Benjamin_Boenisch/breakpilot-pwa
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
This repository has been archived on 2026-02-15. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
main
breakpilot-pwa/backend/klausur/services/module_linker.py
Benjamin Admin 21a844cb8a fix: Restore all files lost during destructive rebase 
A previous `git pull --rebase origin main` dropped 177 local commits,
losing 3400+ files across admin-v2, backend, studio-v2, website,
klausur-service, and many other services. The partial restore attempt
(660295e2) only recovered some files.

This commit restores all missing files from pre-rebase ref 98933f5e
while preserving post-rebase additions (night-scheduler, night-mode UI,
NightModeWidget dashboard integration).

Restored features include:
- AI Module Sidebar (FAB), OCR Labeling, OCR Compare
- GPU Dashboard, RAG Pipeline, Magic Help
- Klausur-Korrektur (8 files), Abitur-Archiv (5+ files)
- Companion, Zeugnisse-Crawler, Screen Flow
- Full backend, studio-v2, website, klausur-service
- All compliance SDKs, agent-core, voice-service
- CI/CD configs, documentation, scripts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-09 09:51:32 +01:00

631 lines
21 KiB
Python
Raw Permalink Blame History

"""
Module Linker Service - Cross-Module Verknuepfungen.
Verknuepft Klausur-Ergebnisse mit anderen BreakPilot-Modulen:
- Notenbuch (School Service)
- Elternabend (Gespraechsvorschlaege)
- Zeugnisse (Notenuebernahme)
- Kalender (Termine)
Privacy:
- Verknuepfungen nutzen doc_tokens (pseudonymisiert)
- Deanonymisierung nur Client-seitig moeglich
"""
import httpx
import os
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime, timedelta
from enum import Enum
# ============================================================================
# DATA CLASSES
# ============================================================================
class LinkType(str, Enum):
"""Typ der Modul-Verknuepfung."""
NOTENBUCH = "notenbuch"
ELTERNABEND = "elternabend"
ZEUGNIS = "zeugnis"
CALENDAR = "calendar"
KLASSENBUCH = "klassenbuch"
class MeetingUrgency(str, Enum):
"""Dringlichkeit eines Elterngespraechs."""
LOW = "niedrig"
MEDIUM = "mittel"
HIGH = "hoch"
@dataclass
class CorrectionResult:
"""Korrektur-Ergebnis (pseudonymisiert)."""
doc_token: str
score: float # Punkte
max_score: float
grade: str # z.B. "2+"
feedback: str
question_results: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class GradeEntry:
"""Notenbuch-Eintrag."""
student_id: str # Im Notenbuch: echte Student-ID
doc_token: str # Aus Klausur: pseudonymisiert
grade: str
points: float
max_points: float
exam_name: str
date: str
@dataclass
class ParentMeetingSuggestion:
"""Vorschlag fuer ein Elterngespraech."""
doc_token: str # Pseudonymisiert
reason: str
urgency: MeetingUrgency
grade: str
subject: str
suggested_topics: List[str] = field(default_factory=list)
@dataclass
class CalendarEvent:
"""Kalender-Eintrag."""
id: str
title: str
description: str
start_time: datetime
end_time: datetime
event_type: str
linked_doc_tokens: List[str] = field(default_factory=list)
@dataclass
class ModuleLink:
"""Verknuepfung zu einem anderen Modul."""
id: str
klausur_session_id: str
link_type: LinkType
target_module: str
target_entity_id: str
target_url: Optional[str] = None
link_metadata: Dict[str, Any] = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.utcnow)
@dataclass
class LinkResult:
"""Ergebnis einer Verknuepfungs-Operation."""
success: bool
link: Optional[ModuleLink] = None
message: str = ""
target_url: Optional[str] = None
# ============================================================================
# MODULE LINKER
# ============================================================================
class ModuleLinker:
"""
Verknuepft Klausur-Ergebnisse mit anderen Modulen.
Beispiel:
linker = ModuleLinker()
# Noten ins Notenbuch uebertragen
result = await linker.link_to_notenbuch(
session_id="session-123",
class_id="class-456",
results=correction_results
)
# Elterngespraeche vorschlagen
suggestions = linker.suggest_elternabend(
results=correction_results,
subject="Mathematik"
)
"""
# Notenschwellen fuer Elterngespraeche
GRADE_THRESHOLDS = {
"1+": 0.95, "1": 0.90, "1-": 0.85,
"2+": 0.80, "2": 0.75, "2-": 0.70,
"3+": 0.65, "3": 0.60, "3-": 0.55,
"4+": 0.50, "4": 0.45, "4-": 0.40,
"5+": 0.33, "5": 0.25, "5-": 0.17,
"6": 0.0
}
# Noten die Gespraeche erfordern
MEETING_TRIGGER_GRADES = ["4", "4-", "5+", "5", "5-", "6"]
def __init__(self):
self.school_service_url = os.getenv(
"SCHOOL_SERVICE_URL",
"http://school-service:8084"
)
self.calendar_service_url = os.getenv(
"CALENDAR_SERVICE_URL",
"http://calendar-service:8085"
)
# =========================================================================
# NOTENBUCH INTEGRATION
# =========================================================================
async def link_to_notenbuch(
self,
session_id: str,
class_id: str,
subject: str,
results: List[CorrectionResult],
exam_name: str,
exam_date: str,
identity_map: Optional[Dict[str, str]] = None
) -> LinkResult:
"""
Uebertraegt Noten ins Notenbuch (School Service).
Args:
session_id: Klausur-Session-ID
class_id: Klassen-ID im School Service
subject: Fach
results: Liste der Korrektur-Ergebnisse
exam_name: Name der Klausur
exam_date: Datum der Klausur
identity_map: Optional: doc_token -> student_id Mapping
Note:
Das identity_map wird nur serverseitig genutzt, wenn der
Lehrer explizit die Verknuepfung freigibt. Normalerweise
bleibt das Mapping Client-seitig.
"""
try:
# Noten-Daten aufbereiten
grades_data = []
for result in results:
grade_entry = {
"doc_token": result.doc_token,
"grade": result.grade,
"points": result.score,
"max_points": result.max_score,
"percentage": result.score / result.max_score if result.max_score > 0 else 0
}
# Falls identity_map vorhanden: Student-ID hinzufuegen
if identity_map and result.doc_token in identity_map:
grade_entry["student_id"] = identity_map[result.doc_token]
grades_data.append(grade_entry)
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{self.school_service_url}/api/classes/{class_id}/exams",
json={
"name": exam_name,
"subject": subject,
"date": exam_date,
"max_points": results[0].max_score if results else 100,
"grades": grades_data,
"klausur_session_id": session_id
}
)
if response.status_code in (200, 201):
data = response.json()
return LinkResult(
success=True,
link=ModuleLink(
id=data.get('id', ''),
klausur_session_id=session_id,
link_type=LinkType.NOTENBUCH,
target_module="school",
target_entity_id=data.get('id', ''),
target_url=f"/app?module=school&class={class_id}&exam={data.get('id')}"
),
message=f"Noten erfolgreich uebertragen ({len(results)} Eintraege)",
target_url=f"/app?module=school&class={class_id}"
)
return LinkResult(
success=False,
message=f"Fehler beim Uebertragen: {response.status_code}"
)
except Exception as e:
return LinkResult(
success=False,
message=f"Verbindungsfehler: {str(e)}"
)
# =========================================================================
# ELTERNABEND VORSCHLAEGE
# =========================================================================
def suggest_elternabend(
self,
results: List[CorrectionResult],
subject: str,
threshold_grade: str = "4"
) -> List[ParentMeetingSuggestion]:
"""
Schlaegt Elterngespraeche fuer schwache Schueler vor.
Args:
results: Liste der Korrektur-Ergebnisse
subject: Fach
threshold_grade: Ab dieser Note wird ein Gespraech vorgeschlagen
Returns:
Liste von Gespraechs-Vorschlaegen (pseudonymisiert)
"""
suggestions = []
threshold_idx = list(self.GRADE_THRESHOLDS.keys()).index(threshold_grade) \
if threshold_grade in self.GRADE_THRESHOLDS else 9
for result in results:
# Pruefe ob Note Gespraech erfordert
if result.grade in self.MEETING_TRIGGER_GRADES:
urgency = self._determine_urgency(result.grade)
topics = self._generate_meeting_topics(result, subject)
suggestions.append(ParentMeetingSuggestion(
doc_token=result.doc_token,
reason=f"Note {result.grade} in {subject}",
urgency=urgency,
grade=result.grade,
subject=subject,
suggested_topics=topics
))
# Nach Dringlichkeit sortieren
urgency_order = {
MeetingUrgency.HIGH: 0,
MeetingUrgency.MEDIUM: 1,
MeetingUrgency.LOW: 2
}
suggestions.sort(key=lambda s: urgency_order[s.urgency])
return suggestions
def _determine_urgency(self, grade: str) -> MeetingUrgency:
"""Bestimmt die Dringlichkeit basierend auf der Note."""
if grade in ["5-", "6"]:
return MeetingUrgency.HIGH
elif grade in ["5", "5+"]:
return MeetingUrgency.MEDIUM
else:
return MeetingUrgency.LOW
def _generate_meeting_topics(
self,
result: CorrectionResult,
subject: str
) -> List[str]:
"""Generiert Gespraechsthemen basierend auf den Ergebnissen."""
topics = []
# Allgemeine Themen
topics.append(f"Leistungsstand in {subject}")
# Basierend auf Feedback
if "Verstaendnis" in result.feedback.lower() or "grundlagen" in result.feedback.lower():
topics.append("Grundlagenverstaendnis foerdern")
if "uebung" in result.feedback.lower():
topics.append("Zusaetzliche Uebungsmoeglichkeiten")
# Basierend auf Aufgaben-Ergebnissen
if result.question_results:
weak_areas = []
for qr in result.question_results:
if qr.get('points_awarded', 0) / qr.get('max_points', 1) < 0.5:
weak_areas.append(qr.get('question_text', ''))
if weak_areas:
topics.append("Gezielte Foerderung in Schwachstellen")
# Standard-Themen
if not topics or len(topics) < 3:
topics.extend([
"Lernstrategien besprechen",
"Unterstuetzungsmoeglichkeiten zu Hause",
"Nachhilfe-Optionen"
])
return topics[:5] # Max 5 Themen
async def create_elternabend_link(
self,
session_id: str,
suggestions: List[ParentMeetingSuggestion],
teacher_id: str
) -> LinkResult:
"""Erstellt Verknuepfungen zum Elternabend-Modul."""
# TODO: Integration mit Elternabend-Modul
# Vorerst nur Metadaten speichern
return LinkResult(
success=True,
link=ModuleLink(
id=f"elternabend-{session_id}",
klausur_session_id=session_id,
link_type=LinkType.ELTERNABEND,
target_module="elternabend",
target_entity_id="",
link_metadata={
"suggestion_count": len(suggestions),
"high_urgency_count": sum(
1 for s in suggestions if s.urgency == MeetingUrgency.HIGH
)
}
),
message=f"{len(suggestions)} Elterngespraeche vorgeschlagen",
target_url="/app?module=elternabend"
)
# =========================================================================
# ZEUGNIS INTEGRATION
# =========================================================================
async def update_zeugnis(
self,
class_id: str,
subject: str,
grades: Dict[str, str],
exam_weight: float = 1.0
) -> LinkResult:
"""
Aktualisiert Zeugnis-Aggregation mit neuen Noten.
Args:
class_id: Klassen-ID
subject: Fach
grades: doc_token -> Note Mapping
exam_weight: Gewichtung der Klausur (Standard: 1.0)
"""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{self.school_service_url}/api/classes/{class_id}/grades/aggregate",
json={
"subject": subject,
"grades": grades,
"weight": exam_weight,
"type": "klausur"
}
)
if response.status_code in (200, 201):
return LinkResult(
success=True,
message="Zeugnis-Daten aktualisiert",
target_url=f"/app?module=school&class={class_id}&tab=certificates"
)
return LinkResult(
success=False,
message=f"Fehler: {response.status_code}"
)
except Exception as e:
return LinkResult(
success=False,
message=f"Verbindungsfehler: {str(e)}"
)
# =========================================================================
# KALENDER INTEGRATION
# =========================================================================
async def create_calendar_events(
self,
teacher_id: str,
suggestions: List[ParentMeetingSuggestion],
default_duration_minutes: int = 30
) -> List[CalendarEvent]:
"""
Erstellt Kalender-Eintraege fuer Elterngespraeche.
Args:
teacher_id: ID des Lehrers
suggestions: Liste der Gespraechs-Vorschlaege
default_duration_minutes: Standard-Dauer pro Gespraech
"""
events = []
# Zeitslots generieren (ab naechster Woche, nachmittags)
start_date = datetime.now() + timedelta(days=7 - datetime.now().weekday())
start_date = start_date.replace(hour=14, minute=0, second=0, microsecond=0)
slot_index = 0
for suggestion in suggestions:
# Zeitslot berechnen
event_start = start_date + timedelta(minutes=slot_index * default_duration_minutes)
event_end = event_start + timedelta(minutes=default_duration_minutes)
# Naechster Tag wenn nach 18 Uhr
if event_start.hour >= 18:
start_date += timedelta(days=1)
start_date = start_date.replace(hour=14)
slot_index = 0
event_start = start_date
event_end = event_start + timedelta(minutes=default_duration_minutes)
event = CalendarEvent(
id=f"meeting-{suggestion.doc_token[:8]}",
title=f"Elterngespraech ({suggestion.grade})",
description=f"Anlass: {suggestion.reason}\n\nThemen:\n" +
"\n".join(f"- {t}" for t in suggestion.suggested_topics),
start_time=event_start,
end_time=event_end,
event_type="parent_meeting",
linked_doc_tokens=[suggestion.doc_token]
)
events.append(event)
slot_index += 1
# An Kalender-Service senden
try:
async with httpx.AsyncClient(timeout=10.0) as client:
for event in events:
await client.post(
f"{self.calendar_service_url}/api/events",
json={
"teacher_id": teacher_id,
"title": event.title,
"description": event.description,
"start": event.start_time.isoformat(),
"end": event.end_time.isoformat(),
"type": event.event_type,
"metadata": {
"doc_tokens": event.linked_doc_tokens
}
}
)
except Exception as e:
print(f"[ModuleLinker] Calendar service error: {e}")
return events
# =========================================================================
# STATISTIKEN
# =========================================================================
def calculate_grade_statistics(
self,
results: List[CorrectionResult]
) -> Dict[str, Any]:
"""
Berechnet Notenstatistiken.
Returns:
Dict mit Durchschnitt, Verteilung, Median, etc.
"""
if not results:
return {}
# Notenwerte (fuer Durchschnitt)
grade_values = {
"1+": 0.7, "1": 1.0, "1-": 1.3,
"2+": 1.7, "2": 2.0, "2-": 2.3,
"3+": 2.7, "3": 3.0, "3-": 3.3,
"4+": 3.7, "4": 4.0, "4-": 4.3,
"5+": 4.7, "5": 5.0, "5-": 5.3,
"6": 6.0
}
# Noten sammeln
grades = [r.grade for r in results]
points = [r.score for r in results]
max_points = results[0].max_score if results else 100
# Durchschnitt berechnen
numeric_grades = [grade_values.get(g, 4.0) for g in grades]
avg_grade = sum(numeric_grades) / len(numeric_grades)
# Notenverteilung
distribution = {}
for grade in grades:
distribution[grade] = distribution.get(grade, 0) + 1
# Prozent-Verteilung
percent_distribution = {
"sehr gut (1)": sum(1 for g in grades if g.startswith("1")),
"gut (2)": sum(1 for g in grades if g.startswith("2")),
"befriedigend (3)": sum(1 for g in grades if g.startswith("3")),
"ausreichend (4)": sum(1 for g in grades if g.startswith("4")),
"mangelhaft (5)": sum(1 for g in grades if g.startswith("5")),
"ungenuegend (6)": sum(1 for g in grades if g == "6")
}
return {
"count": len(results),
"average_grade": round(avg_grade, 2),
"average_grade_display": self._numeric_to_grade(avg_grade),
"average_points": round(sum(points) / len(points), 1),
"max_points": max_points,
"average_percent": round((sum(points) / len(points) / max_points) * 100, 1),
"best_grade": min(grades, key=lambda g: grade_values.get(g, 6)),
"worst_grade": max(grades, key=lambda g: grade_values.get(g, 0)),
"median_grade": self._calculate_median_grade(grades),
"distribution": distribution,
"percent_distribution": percent_distribution,
"passing_count": sum(1 for g in grades if not g.startswith("5") and g != "6"),
"failing_count": sum(1 for g in grades if g.startswith("5") or g == "6")
}
def _numeric_to_grade(self, value: float) -> str:
"""Konvertiert Notenwert zu Note."""
if value <= 1.15:
return "1+"
elif value <= 1.5:
return "1"
elif value <= 1.85:
return "1-"
elif value <= 2.15:
return "2+"
elif value <= 2.5:
return "2"
elif value <= 2.85:
return "2-"
elif value <= 3.15:
return "3+"
elif value <= 3.5:
return "3"
elif value <= 3.85:
return "3-"
elif value <= 4.15:
return "4+"
elif value <= 4.5:
return "4"
elif value <= 4.85:
return "4-"
elif value <= 5.15:
return "5+"
elif value <= 5.5:
return "5"
elif value <= 5.85:
return "5-"
else:
return "6"
def _calculate_median_grade(self, grades: List[str]) -> str:
"""Berechnet die Median-Note."""
grade_values = {
"1+": 0.7, "1": 1.0, "1-": 1.3,
"2+": 1.7, "2": 2.0, "2-": 2.3,
"3+": 2.7, "3": 3.0, "3-": 3.3,
"4+": 3.7, "4": 4.0, "4-": 4.3,
"5+": 4.7, "5": 5.0, "5-": 5.3,
"6": 6.0
}
numeric = sorted([grade_values.get(g, 4.0) for g in grades])
n = len(numeric)
if n % 2 == 0:
median = (numeric[n // 2 - 1] + numeric[n // 2]) / 2
else:
median = numeric[n // 2]
return self._numeric_to_grade(median)
# Singleton
_module_linker: Optional[ModuleLinker] = None
def get_module_linker() -> ModuleLinker:
"""Gibt die Singleton-Instanz des ModuleLinkers zurueck."""
global _module_linker
if _module_linker is None:
_module_linker = ModuleLinker()
return _module_linker
Reference in New Issue View Git Blame Copy Permalink