Files
Benjamin Boenisch 5a31f52310 Initial commit: breakpilot-lehrer - Lehrer KI Platform
Services: Admin-Lehrer, Backend-Lehrer, Studio v2, Website,
Klausur-Service, School-Service, Voice-Service, Geo-Service,
BreakPilot Drive, Agent-Core

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-11 23:47:26 +01:00

677 lines
22 KiB
Python

"""
Antizipations-Engine fuer proaktive Vorschlaege (Phase 8b).
Die Engine sammelt Signale aus verschiedenen Quellen und generiert
kontextbasierte Vorschlaege fuer Lehrer basierend auf definierten Regeln.
Architektur:
1. SignalCollector - Sammelt Inputs (Zeit, Nutzung, Events)
2. RuleEngine - Evaluiert Regeln gegen Signale
3. SuggestionGenerator - Generiert priorisierte Vorschlaege
"""
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from enum import Enum
import logging
logger = logging.getLogger(__name__)
# ==================== Enums & Types ====================
class SuggestionTone(str, Enum):
"""Ton/Dringlichkeit eines Vorschlags."""
HINT = "hint" # Sanfter Hinweis
SUGGESTION = "suggestion" # Aktiver Vorschlag
REMINDER = "reminder" # Erinnerung
URGENT = "urgent" # Dringend
class ContextType(str, Enum):
"""Typ eines aktiven Kontexts."""
EVENT_WINDOW = "event_window" # Event steht bevor
ROUTINE = "routine" # Routine heute
PHASE = "phase" # Makro-Phase bedingt
TIME = "time" # Zeitbasiert (Ferien, Wochenende)
@dataclass
class Signal:
"""Ein einzelnes Signal aus einer Quelle."""
name: str
value: Any
source: str # "calendar", "usage", "events", "routines"
@dataclass
class ActiveContext:
"""Ein aktiver Kontext der Vorschlaege beeinflusst."""
id: str
context_type: ContextType
label: str
data: Dict[str, Any] = field(default_factory=dict)
@dataclass
class Suggestion:
"""Ein generierter Vorschlag."""
id: str
title: str
description: str
tone: SuggestionTone
action_url: Optional[str] = None
badge: Optional[str] = None # z.B. "in 7 Tagen"
priority: int = 50 # 0-100, hoeher = wichtiger
rule_id: str = ""
icon: str = "lightbulb"
@dataclass
class Signals:
"""Container fuer alle gesammelten Signale."""
# Zeit/Kalender
current_week: int = 1
weeks_since_start: int = 0
is_weekend: bool = False
is_before_holidays: bool = False
days_until_holidays: int = 999
# Makro-Phase
macro_phase: str = "onboarding"
onboarding_completed: bool = False
# Produktnutzung
classes_count: int = 0
has_classes: bool = False
has_schedule: bool = False
# Events
exams_scheduled_count: int = 0
exams_in_7_days: List[Dict] = field(default_factory=list)
exams_past_ungraded: List[Dict] = field(default_factory=list)
upcoming_events: List[Dict] = field(default_factory=list)
trips_in_30_days: List[Dict] = field(default_factory=list)
parent_evenings_soon: List[Dict] = field(default_factory=list)
# Routinen
routines_today: List[Dict] = field(default_factory=list)
has_conference_today: bool = False
# Statistiken (aus Analytics)
corrections_pending: int = 0
grades_completion_ratio: float = 0.0
# ==================== Signal Collector ====================
class SignalCollector:
"""
Sammelt Signale aus verschiedenen Quellen.
Quellen:
- TeacherContext (Makro-Phase, Schuljahr)
- SchoolyearEvents (Klausuren, Elternabende, etc.)
- RecurringRoutines (Konferenzen heute)
- Zeit/Kalender (Wochenende, Ferien)
"""
def __init__(self, db_session=None):
self.db = db_session
def collect(self, teacher_id: str) -> Signals:
"""Sammelt alle Signale fuer einen Lehrer."""
signals = Signals()
# Zeit-Signale
self._collect_time_signals(signals)
if self.db:
# Kontext-Signale
self._collect_context_signals(signals, teacher_id)
# Event-Signale
self._collect_event_signals(signals, teacher_id)
# Routine-Signale
self._collect_routine_signals(signals, teacher_id)
return signals
def _collect_time_signals(self, signals: Signals):
"""Sammelt zeitbasierte Signale."""
now = datetime.utcnow()
signals.is_weekend = now.weekday() >= 5
# TODO: Ferien-Kalender pro Bundesland integrieren
# Fuer jetzt: Dummy-Werte
signals.is_before_holidays = False
signals.days_until_holidays = 999
def _collect_context_signals(self, signals: Signals, teacher_id: str):
"""Sammelt Signale aus dem Teacher-Kontext."""
from .repository import TeacherContextRepository
try:
repo = TeacherContextRepository(self.db)
context = repo.get_or_create(teacher_id)
signals.macro_phase = context.macro_phase.value
signals.current_week = context.current_week or 1
signals.onboarding_completed = context.onboarding_completed
signals.has_classes = context.has_classes
signals.has_schedule = context.has_schedule
signals.classes_count = 1 if context.has_classes else 0
# Wochen seit Schuljahresstart berechnen
if context.schoolyear_start:
delta = datetime.utcnow() - context.schoolyear_start
signals.weeks_since_start = max(0, delta.days // 7)
signals.is_before_holidays = context.is_before_holidays
except Exception as e:
logger.warning(f"Failed to collect context signals: {e}")
def _collect_event_signals(self, signals: Signals, teacher_id: str):
"""Sammelt Signale aus Events."""
from .repository import SchoolyearEventRepository
try:
repo = SchoolyearEventRepository(self.db)
now = datetime.utcnow()
# Alle anstehenden Events (30 Tage)
upcoming = repo.get_upcoming(teacher_id, days=30, limit=20)
signals.upcoming_events = [repo.to_dict(e) for e in upcoming]
# Klausuren in den naechsten 7 Tagen
seven_days = now + timedelta(days=7)
signals.exams_in_7_days = [
repo.to_dict(e) for e in upcoming
if e.event_type.value == "exam" and e.start_date <= seven_days
]
signals.exams_scheduled_count = len([
e for e in upcoming if e.event_type.value == "exam"
])
# Klassenfahrten in 30 Tagen
signals.trips_in_30_days = [
repo.to_dict(e) for e in upcoming
if e.event_type.value == "trip"
]
# Elternabende bald
signals.parent_evenings_soon = [
repo.to_dict(e) for e in upcoming
if e.event_type.value in ("parent_evening", "parent_consultation")
]
except Exception as e:
logger.warning(f"Failed to collect event signals: {e}")
def _collect_routine_signals(self, signals: Signals, teacher_id: str):
"""Sammelt Signale aus Routinen."""
from .repository import RecurringRoutineRepository
try:
repo = RecurringRoutineRepository(self.db)
today_routines = repo.get_today(teacher_id)
signals.routines_today = [repo.to_dict(r) for r in today_routines]
signals.has_conference_today = any(
r.routine_type.value in ("teacher_conference", "subject_conference")
for r in today_routines
)
except Exception as e:
logger.warning(f"Failed to collect routine signals: {e}")
# ==================== Rule Engine ====================
@dataclass
class Rule:
"""Eine Regel die Signale zu Vorschlaegen mappt."""
id: str
name: str
description: str
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
"""Evaluiert die Regel und gibt einen Vorschlag zurueck oder None."""
raise NotImplementedError()
class R01_CreateClasses(Rule):
"""Klassen anlegen wenn noch keine vorhanden."""
def __init__(self):
super().__init__(
id="R01",
name="Klassen anlegen",
description="Empfiehlt Klassen anzulegen bei neuem Lehrer"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.macro_phase == "onboarding" and not signals.has_classes:
return Suggestion(
id="suggest_create_classes",
title="Klassen anlegen",
description="Legen Sie Ihre Klassen an, um den vollen Funktionsumfang zu nutzen.",
tone=SuggestionTone.HINT,
priority=90,
rule_id=self.id,
icon="group_add",
action_url="/classes/new",
)
return None
class R02_PrepareRubric(Rule):
"""Erwartungshorizont erstellen wenn Klausur in 7 Tagen."""
def __init__(self):
super().__init__(
id="R02",
name="Erwartungshorizont",
description="Empfiehlt Erwartungshorizont vor Klausur"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.exams_in_7_days:
exam = signals.exams_in_7_days[0]
# Pruefen ob Vorbereitung noch nicht erledigt
if not exam.get("preparation_done", False):
days = 7 # Vereinfacht
return Suggestion(
id=f"suggest_rubric_{exam['id'][:8]}",
title="Erwartungshorizont erstellen",
description=f"Klausur '{exam['title']}' steht bevor.",
tone=SuggestionTone.SUGGESTION,
badge=f"in {days} Tagen",
priority=80,
rule_id=self.id,
icon="assignment",
action_url=f"/exams/{exam['id']}/rubric",
)
return None
class R03_StartCorrection(Rule):
"""Korrektur starten nach Klausur."""
def __init__(self):
super().__init__(
id="R03",
name="Korrektur starten",
description="Empfiehlt Korrektur nach Klausur"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.exams_past_ungraded:
exam = signals.exams_past_ungraded[0]
return Suggestion(
id=f"suggest_correction_{exam['id'][:8]}",
title="Korrektur-Setup starten",
description=f"Klausur '{exam['title']}' ist geschrieben.",
tone=SuggestionTone.HINT,
badge="bereit",
priority=75,
rule_id=self.id,
icon="rate_review",
action_url=f"/exams/{exam['id']}/correct",
)
return None
class R05_PrepareAgenda(Rule):
"""Agenda vorbereiten wenn Konferenz heute."""
def __init__(self):
super().__init__(
id="R05",
name="Konferenz-Agenda",
description="Empfiehlt Agenda wenn Konferenz heute"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.has_conference_today:
# Finde die Konferenz
conf = next(
(r for r in signals.routines_today
if r.get("routine_type") in ("teacher_conference", "subject_conference")),
None
)
if conf:
return Suggestion(
id="suggest_agenda",
title="Konferenz-Agenda vorbereiten",
description=f"{conf.get('title', 'Konferenz')} heute.",
tone=SuggestionTone.SUGGESTION,
badge="heute",
priority=70,
rule_id=self.id,
icon="event_note",
)
return None
class R07_PlanFirstExam(Rule):
"""Erste Klausur planen nach 4 Wochen."""
def __init__(self):
super().__init__(
id="R07",
name="Erste Arbeit planen",
description="Empfiehlt erste Klausur nach Anlaufphase"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if (signals.weeks_since_start >= 4 and
signals.exams_scheduled_count == 0 and
signals.has_classes):
return Suggestion(
id="suggest_first_exam",
title="Erste Klassenarbeit planen",
description="Nach 4 Wochen Unterricht ist ein guter Zeitpunkt fuer die erste Leistungsueberpruefung.",
tone=SuggestionTone.SUGGESTION,
priority=60,
rule_id=self.id,
icon="quiz",
action_url="/exams/new",
)
return None
class R08_CorrectionMode(Rule):
"""Korrekturmodus vor Ferien."""
def __init__(self):
super().__init__(
id="R08",
name="Ferien-Korrekturmodus",
description="Empfiehlt Korrekturen vor Ferien"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.is_before_holidays and signals.corrections_pending > 0:
return Suggestion(
id="suggest_correction_mode",
title="Ferien-Korrekturmodus",
description=f"{signals.corrections_pending} Korrekturen noch offen vor den Ferien.",
tone=SuggestionTone.REMINDER,
badge=f"{signals.days_until_holidays}d bis Ferien",
priority=65,
rule_id=self.id,
icon="grading",
)
return None
class R09_TripChecklist(Rule):
"""Klassenfahrt-Checkliste wenn Fahrt in 30 Tagen."""
def __init__(self):
super().__init__(
id="R09",
name="Klassenfahrt-Checkliste",
description="Empfiehlt Checkliste vor Klassenfahrt"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.trips_in_30_days:
trip = signals.trips_in_30_days[0]
return Suggestion(
id=f"suggest_trip_{trip['id'][:8]}",
title="Klassenfahrt-Checkliste",
description=f"'{trip['title']}' steht bevor.",
tone=SuggestionTone.SUGGESTION,
badge="in 30 Tagen",
priority=55,
rule_id=self.id,
icon="luggage",
action_url=f"/trips/{trip['id']}/checklist",
)
return None
class R10_CompleteGrades(Rule):
"""Noten vervollstaendigen vor Halbjahresende."""
def __init__(self):
super().__init__(
id="R10",
name="Noten vervollstaendigen",
description="Empfiehlt Noten vor Notenschluss"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if (signals.macro_phase in ("halbjahresabschluss", "jahresabschluss") and
signals.grades_completion_ratio < 0.8):
pct = int(signals.grades_completion_ratio * 100)
return Suggestion(
id="suggest_complete_grades",
title="Noten vervollstaendigen",
description=f"Nur {pct}% der Noten eingetragen. Notenschluss naht!",
tone=SuggestionTone.REMINDER,
priority=85,
rule_id=self.id,
icon="calculate",
action_url="/grades",
)
return None
class R11_SetupSchedule(Rule):
"""Stundenplan einrichten wenn noch nicht vorhanden."""
def __init__(self):
super().__init__(
id="R11",
name="Stundenplan einrichten",
description="Empfiehlt Stundenplan-Setup"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.macro_phase == "onboarding" and not signals.has_schedule:
return Suggestion(
id="suggest_setup_schedule",
title="Stundenplan einrichten",
description="Richten Sie Ihren Stundenplan ein fuer personalisierte Vorschlaege.",
tone=SuggestionTone.HINT,
priority=85,
rule_id=self.id,
icon="calendar_month",
action_url="/schedule/setup",
)
return None
class R12_ParentEvening(Rule):
"""Elternabend vorbereiten."""
def __init__(self):
super().__init__(
id="R12",
name="Elternabend vorbereiten",
description="Empfiehlt Vorbereitung vor Elternabend"
)
def evaluate(self, signals: Signals) -> Optional[Suggestion]:
if signals.parent_evenings_soon:
event = signals.parent_evenings_soon[0]
return Suggestion(
id=f"suggest_parent_{event['id'][:8]}",
title="Elternabend vorbereiten",
description=f"'{event['title']}' steht bevor.",
tone=SuggestionTone.SUGGESTION,
badge="bald",
priority=65,
rule_id=self.id,
icon="family_restroom",
action_url=f"/events/{event['id']}/prepare",
)
return None
class RuleEngine:
"""
Evaluiert alle Regeln gegen die gesammelten Signale.
"""
def __init__(self):
self.rules: List[Rule] = [
R01_CreateClasses(),
R02_PrepareRubric(),
R03_StartCorrection(),
R05_PrepareAgenda(),
R07_PlanFirstExam(),
R08_CorrectionMode(),
R09_TripChecklist(),
R10_CompleteGrades(),
R11_SetupSchedule(),
R12_ParentEvening(),
]
def evaluate(self, signals: Signals) -> List[Suggestion]:
"""Evaluiert alle Regeln und gibt passende Vorschlaege zurueck."""
suggestions = []
for rule in self.rules:
try:
suggestion = rule.evaluate(signals)
if suggestion:
suggestions.append(suggestion)
except Exception as e:
logger.warning(f"Rule {rule.id} failed: {e}")
# Nach Prioritaet sortieren (hoechste zuerst)
suggestions.sort(key=lambda s: s.priority, reverse=True)
return suggestions
# ==================== Suggestion Generator ====================
class SuggestionGenerator:
"""
Hauptklasse die Signale sammelt, Regeln evaluiert und
Vorschlaege generiert.
"""
def __init__(self, db_session=None):
self.collector = SignalCollector(db_session)
self.rule_engine = RuleEngine()
def generate(self, teacher_id: str, limit: int = 5) -> Dict[str, Any]:
"""
Generiert Vorschlaege fuer einen Lehrer.
Returns:
{
"active_contexts": [...],
"suggestions": [...],
"signals_summary": {...}
}
"""
# 1. Signale sammeln
signals = self.collector.collect(teacher_id)
# 2. Regeln evaluieren
all_suggestions = self.rule_engine.evaluate(signals)
# 3. Aktive Kontexte bestimmen
active_contexts = self._determine_active_contexts(signals)
# 4. Top N Vorschlaege
top_suggestions = all_suggestions[:limit]
return {
"active_contexts": [
{
"id": ctx.id,
"type": ctx.context_type.value,
"label": ctx.label,
}
for ctx in active_contexts
],
"suggestions": [
{
"id": s.id,
"title": s.title,
"description": s.description,
"tone": s.tone.value,
"badge": s.badge,
"priority": s.priority,
"icon": s.icon,
"action_url": s.action_url,
}
for s in top_suggestions
],
"signals_summary": {
"macro_phase": signals.macro_phase,
"current_week": signals.current_week,
"has_classes": signals.has_classes,
"exams_soon": len(signals.exams_in_7_days),
"routines_today": len(signals.routines_today),
},
"total_suggestions": len(all_suggestions),
}
def _determine_active_contexts(self, signals: Signals) -> List[ActiveContext]:
"""Bestimmt die aktiven Kontexte basierend auf Signalen."""
contexts = []
# Event-Kontexte
if signals.exams_in_7_days:
contexts.append(ActiveContext(
id="EXAM_IN_7_DAYS",
context_type=ContextType.EVENT_WINDOW,
label="Klausur in 7 Tagen",
))
if signals.trips_in_30_days:
contexts.append(ActiveContext(
id="TRIP_UPCOMING",
context_type=ContextType.EVENT_WINDOW,
label="Klassenfahrt geplant",
))
# Routine-Kontexte
if signals.has_conference_today:
contexts.append(ActiveContext(
id="CONFERENCE_TODAY",
context_type=ContextType.ROUTINE,
label="Konferenz heute",
))
# Zeit-Kontexte
if signals.is_weekend:
contexts.append(ActiveContext(
id="WEEKEND",
context_type=ContextType.TIME,
label="Wochenende",
))
if signals.is_before_holidays:
contexts.append(ActiveContext(
id="BEFORE_HOLIDAYS",
context_type=ContextType.TIME,
label="Vor den Ferien",
))
# Phase-Kontexte
if signals.macro_phase == "onboarding":
contexts.append(ActiveContext(
id="ONBOARDING",
context_type=ContextType.PHASE,
label="Einrichtung",
))
elif signals.macro_phase in ("halbjahresabschluss", "jahresabschluss"):
contexts.append(ActiveContext(
id="GRADE_PERIOD",
context_type=ContextType.PHASE,
label="Notenphase",
))
return contexts