"""
Klausur-Korrektur API - REST API für Abitur-Klausur-Korrektur.
Zwei Modi:
- Modus A: Landes-Abitur Niedersachsen (NiBiS-Aufgaben, rechtlich geklärt)
- Modus B: Vorabitur (Lehrer-erstellte Klausuren mit Rights-Gate)
Features:
- Multi-Kriterien-Bewertung (Rechtschreibung, Grammatik, Inhalt, Struktur, Stil)
- Rights-Gate für Textquellen-Verifikation
- KI-generierter Erwartungshorizont
- Gutachten-Generierung
- 15-Punkte-Notensystem (Abitur)
- Erst-/Zweitprüfer-Workflow
- Fairness-Vergleich
"""
import logging
import uuid
import os
from datetime import datetime
from typing import List, Dict, Any, Optional
from enum import Enum
from pathlib import Path
from dataclasses import dataclass, field, asdict
from fastapi import APIRouter, HTTPException, UploadFile, File, Form, BackgroundTasks
from fastapi.responses import Response, FileResponse
from pydantic import BaseModel, Field
# FileProcessor requires OpenCV with libGL - make optional for CI
try:
from services.file_processor import FileProcessor
_ocr_available = True
except (ImportError, OSError):
FileProcessor = None # type: ignore
_ocr_available = False
# PDF service requires WeasyPrint with system libraries - make optional for CI
try:
from services.pdf_service import PDFService
_pdf_available = True
except (ImportError, OSError):
PDFService = None # type: ignore
_pdf_available = False
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/klausur-korrektur",
tags=["klausur-korrektur"],
)
# Upload directory
UPLOAD_DIR = Path("/tmp/klausur-korrektur")
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
# ============================================================================
# Enums
# ============================================================================
class KlausurModus(str, Enum):
"""Klausur-Modus."""
LANDES_ABITUR = "landes_abitur" # NiBiS-Aufgaben
VORABITUR = "vorabitur" # Lehrer-erstellt
class TextSourceType(str, Enum):
"""Art der Textquelle."""
NIBIS = "nibis" # Offizielle NiBiS-Aufgabe
EIGENTEXT = "eigentext" # Eigener Text
FREMDTEXT = "fremdtext" # Fremder Text (erfordert Rights-Gate)
class TextSourceStatus(str, Enum):
"""Status der Textquellen-Verifikation."""
PENDING = "pending"
VERIFIED = "verified"
REJECTED = "rejected"
class StudentKlausurStatus(str, Enum):
"""Status einer Schüler-Klausur."""
UPLOADED = "uploaded"
OCR_PROCESSING = "ocr_processing"
OCR_COMPLETE = "ocr_complete"
ANALYZING = "analyzing"
FIRST_EXAMINER = "first_examiner"
SECOND_EXAMINER = "second_examiner"
COMPLETED = "completed"
ERROR = "error"
class KlausurStatus(str, Enum):
"""Status einer Klausur."""
DRAFT = "draft"
TEXT_SOURCE = "text_source"
RIGHTS_GATE = "rights_gate"
ERWARTUNGSHORIZONT = "erwartungshorizont"
COLLECTING = "collecting"
CORRECTING = "correcting"
COMPLETED = "completed"
class Anforderungsbereich(int, Enum):
"""Anforderungsbereich nach KMK."""
I = 1 # Reproduktion
II = 2 # Reorganisation und Transfer
III = 3 # Reflexion und Problemlösung
# ============================================================================
# 15-Punkte-Notenschlüssel
# ============================================================================
GRADE_THRESHOLDS = {
15: 95, # 1+
14: 90, # 1
13: 85, # 1-
12: 80, # 2+
11: 75, # 2
10: 70, # 2-
9: 65, # 3+
8: 60, # 3
7: 55, # 3-
6: 50, # 4+
5: 45, # 4
4: 40, # 4-
3: 33, # 5+
2: 27, # 5
1: 20, # 5-
0: 0, # 6
}
GRADE_LABELS = {
15: "1+ (sehr gut plus)",
14: "1 (sehr gut)",
13: "1- (sehr gut minus)",
12: "2+ (gut plus)",
11: "2 (gut)",
10: "2- (gut minus)",
9: "3+ (befriedigend plus)",
8: "3 (befriedigend)",
7: "3- (befriedigend minus)",
6: "4+ (ausreichend plus)",
5: "4 (ausreichend)",
4: "4- (ausreichend minus)",
3: "5+ (mangelhaft plus)",
2: "5 (mangelhaft)",
1: "5- (mangelhaft minus)",
0: "6 (ungenügend)",
}
# Standard-Bewertungskriterien
DEFAULT_CRITERIA = {
"rechtschreibung": {"weight": 0.15, "max": 100, "label": "Rechtschreibung"},
"grammatik": {"weight": 0.15, "max": 100, "label": "Grammatik"},
"inhalt": {"weight": 0.40, "max": 100, "label": "Inhalt"},
"struktur": {"weight": 0.15, "max": 100, "label": "Struktur"},
"stil": {"weight": 0.15, "max": 100, "label": "Stil"},
}
# Operatoren nach Niedersachsen KC
OPERATORS = {
"I": [
{"name": "nennen", "description": "Informationen ohne Erläuterung wiedergeben"},
{"name": "beschreiben", "description": "Sachverhalte in eigenen Worten wiedergeben"},
{"name": "wiedergeben", "description": "Inhalte in eigenen Worten darstellen"},
{"name": "zusammenfassen", "description": "Wesentliche Aspekte komprimiert darstellen"},
],
"II": [
{"name": "analysieren", "description": "Materialien systematisch untersuchen"},
{"name": "erklären", "description": "Sachverhalte verständlich machen"},
{"name": "erläutern", "description": "Sachverhalte mit Beispielen veranschaulichen"},
{"name": "vergleichen", "description": "Gemeinsamkeiten und Unterschiede herausarbeiten"},
{"name": "einordnen", "description": "In einen Zusammenhang stellen"},
{"name": "charakterisieren", "description": "Wesenszüge herausarbeiten"},
],
"III": [
{"name": "beurteilen", "description": "Aussagen an Kriterien messen und ein Urteil fällen"},
{"name": "bewerten", "description": "Eine eigene Position mit Begründung einnehmen"},
{"name": "erörtern", "description": "Ein Thema multiperspektivisch diskutieren"},
{"name": "Stellung nehmen", "description": "Eigene Position argumentativ vertreten"},
{"name": "gestalten", "description": "Kreativ-produktive Texte erstellen"},
{"name": "entwerfen", "description": "Konzepte entwickeln"},
],
}
# ============================================================================
# Pydantic Models für API Requests/Responses
# ============================================================================
class TextSourceCreate(BaseModel):
"""Request zum Erstellen einer Textquelle."""
source_type: TextSourceType
title: str
author: str = ""
content: str
nibis_id: Optional[str] = None # Für NiBiS-Aufgaben
license_info: Optional[Dict[str, Any]] = None
class TextSourceResponse(BaseModel):
"""Response für eine Textquelle."""
id: str
source_type: TextSourceType
title: str
author: str
content: str
nibis_id: Optional[str]
license_status: TextSourceStatus
license_info: Optional[Dict[str, Any]]
created_at: datetime
class AufgabeCreate(BaseModel):
"""Request zum Erstellen einer Aufgabe."""
nummer: str
text: str
operator: str
anforderungsbereich: int = Field(ge=1, le=3)
erwartete_leistungen: List[str] = []
punkte: int = Field(ge=0)
class AufgabeResponse(BaseModel):
"""Response für eine Aufgabe."""
id: str
nummer: str
text: str
operator: str
anforderungsbereich: int
erwartete_leistungen: List[str]
punkte: int
class ErwartungshorizontCreate(BaseModel):
"""Request zum Erstellen/Aktualisieren eines Erwartungshorizonts."""
aufgaben: List[AufgabeCreate]
max_points: int = Field(ge=0)
hinweise: str = ""
class ErwartungshorizontResponse(BaseModel):
"""Response für einen Erwartungshorizont."""
id: str
aufgaben: List[AufgabeResponse]
max_points: int
hinweise: str
generated: bool
created_at: datetime
class CriterionScoreUpdate(BaseModel):
"""Update für einen Kriterien-Score."""
score: int = Field(ge=0, le=100)
annotations: List[str] = []
comment: str = ""
class AnnotationCreate(BaseModel):
"""Request zum Erstellen einer Annotation."""
page: int
x: float
y: float
width: float
height: float
text: str
type: str = "comment" # comment, correction, highlight
color: str = "#FF0000"
class AnnotationResponse(BaseModel):
"""Response für eine Annotation."""
id: str
page: int
x: float
y: float
width: float
height: float
text: str
type: str
color: str
created_at: datetime
class GutachtenCreate(BaseModel):
"""Request zum Erstellen/Aktualisieren eines Gutachtens."""
einleitung: str
hauptteil: str
fazit: str
staerken: List[str] = []
schwaechen: List[str] = []
class GutachtenResponse(BaseModel):
"""Response für ein Gutachten."""
id: str
einleitung: str
hauptteil: str
fazit: str
staerken: List[str]
schwaechen: List[str]
generated: bool
edited: bool
created_at: datetime
class ExaminerResultCreate(BaseModel):
"""Request für Prüfer-Ergebnis."""
examiner_id: str
examiner_name: str
grade_points: int = Field(ge=0, le=15)
comment: str = ""
class ExaminerResultResponse(BaseModel):
"""Response für Prüfer-Ergebnis."""
examiner_id: str
examiner_name: str
grade_points: int
comment: str
submitted_at: datetime
class StudentKlausurCreate(BaseModel):
"""Request zum Erstellen einer Schüler-Klausur."""
student_name: str
student_id: Optional[str] = None
class StudentKlausurResponse(BaseModel):
"""Response für eine Schüler-Klausur."""
id: str
student_name: str
student_id: Optional[str]
file_path: Optional[str]
file_name: Optional[str]
ocr_text: Optional[str]
status: StudentKlausurStatus
criteria_scores: Dict[str, Dict[str, Any]]
annotations: List[AnnotationResponse]
gutachten: Optional[GutachtenResponse]
raw_points: int
grade_points: int
grade_label: str
first_examiner: Optional[ExaminerResultResponse]
second_examiner: Optional[ExaminerResultResponse]
created_at: datetime
updated_at: datetime
class KlausurCreate(BaseModel):
"""Request zum Erstellen einer Klausur."""
title: str
subject: str
modus: KlausurModus
year: int = Field(ge=2020, le=2100)
semester: str # Q1, Q2, Q3, Q4
kurs: str = "" # z.B. "Deutsch LK", "Englisch GK"
class_id: Optional[str] = None # ID der zugeordneten Klasse
class KlausurUpdate(BaseModel):
"""Request zum Aktualisieren einer Klausur."""
title: Optional[str] = None
subject: Optional[str] = None
kurs: Optional[str] = None
status: Optional[KlausurStatus] = None
class KlausurResponse(BaseModel):
"""Response für eine Klausur."""
id: str
title: str
subject: str
modus: KlausurModus
year: int
semester: str
kurs: str
class_id: Optional[str]
status: KlausurStatus
text_sources: List[TextSourceResponse]
erwartungshorizont: Optional[ErwartungshorizontResponse]
student_count: int
completed_count: int
average_grade: Optional[float]
created_at: datetime
updated_at: datetime
class NiBiSAufgabe(BaseModel):
"""NiBiS-Aufgabe aus dem Katalog."""
id: str
year: int
subject: str
title: str
type: str # "schriftlich", "mündlich"
text_preview: str
download_url: Optional[str]
class FairnessReport(BaseModel):
"""Fairness-Bericht für eine Klausur."""
klausur_id: str
total_students: int
average_grade: float
grade_distribution: Dict[int, int]
criteria_averages: Dict[str, float]
outliers: List[Dict[str, Any]]
recommendations: List[str]
# ============================================================================
# Internal Data Classes (In-Memory Storage)
# ============================================================================
@dataclass
class TextSource:
"""Textquelle für Klausur."""
id: str
source_type: TextSourceType
title: str
author: str
content: str
nibis_id: Optional[str]
license_status: TextSourceStatus
license_info: Optional[Dict[str, Any]]
created_at: datetime
@dataclass
class Aufgabe:
"""Aufgabe im Erwartungshorizont."""
id: str
nummer: str
text: str
operator: str
anforderungsbereich: int
erwartete_leistungen: List[str]
punkte: int
@dataclass
class Erwartungshorizont:
"""Erwartungshorizont für Klausur."""
id: str
aufgaben: List[Aufgabe]
max_points: int
hinweise: str
generated: bool
created_at: datetime
@dataclass
class Annotation:
"""Overlay-Annotation auf Dokument."""
id: str
page: int
x: float
y: float
width: float
height: float
text: str
type: str
color: str
created_at: datetime
@dataclass
class Gutachten:
"""Gutachten für Schüler-Klausur."""
id: str
einleitung: str
hauptteil: str
fazit: str
staerken: List[str]
schwaechen: List[str]
generated: bool
edited: bool
created_at: datetime
@dataclass
class ExaminerResult:
"""Prüfer-Ergebnis."""
examiner_id: str
examiner_name: str
grade_points: int
comment: str
submitted_at: datetime
@dataclass
class CriterionScore:
"""Bewertung für ein Kriterium."""
score: int
weight: float
annotations: List[str]
comment: str
ai_suggestions: List[str]
@dataclass
class StudentKlausur:
"""Schüler-Klausur."""
id: str
student_name: str
student_id: Optional[str]
file_path: Optional[str]
file_name: Optional[str]
ocr_text: Optional[str]
status: StudentKlausurStatus
criteria_scores: Dict[str, CriterionScore]
annotations: List[Annotation]
gutachten: Optional[Gutachten]
raw_points: int
grade_points: int
first_examiner: Optional[ExaminerResult]
second_examiner: Optional[ExaminerResult]
created_at: datetime
updated_at: datetime
@dataclass
class AbiturKlausur:
"""Abitur-Klausur."""
id: str
title: str
subject: str
modus: KlausurModus
year: int
semester: str
kurs: str
class_id: Optional[str] # ID der zugeordneten Klasse
status: KlausurStatus
text_sources: List[TextSource]
erwartungshorizont: Optional[Erwartungshorizont]
students: List[StudentKlausur]
created_at: datetime
updated_at: datetime
# ============================================================================
# In-Memory Storage
# ============================================================================
_klausuren: Dict[str, AbiturKlausur] = {}
# NiBiS-Katalog (Demo-Daten)
_nibis_katalog: Dict[str, NiBiSAufgabe] = {
"nibis-2024-deutsch-01": NiBiSAufgabe(
id="nibis-2024-deutsch-01",
year=2024,
subject="Deutsch",
title="Abitur 2024 - Aufgabenstellung I (Lyrik)",
type="schriftlich",
text_preview="Analysieren Sie das Gedicht von Else Lasker-Schüler...",
download_url=None
),
"nibis-2024-deutsch-02": NiBiSAufgabe(
id="nibis-2024-deutsch-02",
year=2024,
subject="Deutsch",
title="Abitur 2024 - Aufgabenstellung II (Drama)",
type="schriftlich",
text_preview="Analysieren Sie den Szenenausschnitt aus 'Faust I'...",
download_url=None
),
"nibis-2024-deutsch-03": NiBiSAufgabe(
id="nibis-2024-deutsch-03",
year=2024,
subject="Deutsch",
title="Abitur 2024 - Aufgabenstellung III (Epik)",
type="schriftlich",
text_preview="Analysieren Sie den Romanausschnitt aus 'Der Prozess'...",
download_url=None
),
"nibis-2023-deutsch-01": NiBiSAufgabe(
id="nibis-2023-deutsch-01",
year=2023,
subject="Deutsch",
title="Abitur 2023 - Aufgabenstellung I (Lyrik)",
type="schriftlich",
text_preview="Analysieren Sie das Gedicht 'Mondnacht' von Eichendorff...",
download_url=None
),
"nibis-2024-englisch-01": NiBiSAufgabe(
id="nibis-2024-englisch-01",
year=2024,
subject="Englisch",
title="Abitur 2024 - Reading Comprehension",
type="schriftlich",
text_preview="Read the following article about climate change...",
download_url=None
),
}
# ============================================================================
# Helper Functions
# ============================================================================
def calculate_15_point_grade(percentage: float) -> int:
"""Berechnet 15-Punkte-Note aus Prozent."""
for points, threshold in sorted(GRADE_THRESHOLDS.items(), reverse=True):
if percentage >= threshold:
return points
return 0
def calculate_raw_points(criteria_scores: Dict[str, CriterionScore], max_points: int) -> int:
"""Berechnet Rohpunkte aus Kriterien-Scores."""
if not criteria_scores:
return 0
weighted_sum = 0.0
total_weight = 0.0
for criterion, score in criteria_scores.items():
weighted_sum += score.score * score.weight
total_weight += score.weight
if total_weight == 0:
return 0
percentage = weighted_sum / total_weight
return int(max_points * percentage / 100)
def get_grade_label(grade_points: int) -> str:
"""Gibt Label für Note zurück."""
return GRADE_LABELS.get(grade_points, "unbekannt")
def _to_text_source_response(ts: TextSource) -> TextSourceResponse:
"""Konvertiert TextSource zu Response."""
return TextSourceResponse(
id=ts.id,
source_type=ts.source_type,
title=ts.title,
author=ts.author,
content=ts.content,
nibis_id=ts.nibis_id,
license_status=ts.license_status,
license_info=ts.license_info,
created_at=ts.created_at
)
def _to_aufgabe_response(a: Aufgabe) -> AufgabeResponse:
"""Konvertiert Aufgabe zu Response."""
return AufgabeResponse(
id=a.id,
nummer=a.nummer,
text=a.text,
operator=a.operator,
anforderungsbereich=a.anforderungsbereich,
erwartete_leistungen=a.erwartete_leistungen,
punkte=a.punkte
)
def _to_erwartungshorizont_response(eh: Erwartungshorizont) -> ErwartungshorizontResponse:
"""Konvertiert Erwartungshorizont zu Response."""
return ErwartungshorizontResponse(
id=eh.id,
aufgaben=[_to_aufgabe_response(a) for a in eh.aufgaben],
max_points=eh.max_points,
hinweise=eh.hinweise,
generated=eh.generated,
created_at=eh.created_at
)
def _to_annotation_response(a: Annotation) -> AnnotationResponse:
"""Konvertiert Annotation zu Response."""
return AnnotationResponse(
id=a.id,
page=a.page,
x=a.x,
y=a.y,
width=a.width,
height=a.height,
text=a.text,
type=a.type,
color=a.color,
created_at=a.created_at
)
def _to_gutachten_response(g: Gutachten) -> GutachtenResponse:
"""Konvertiert Gutachten zu Response."""
return GutachtenResponse(
id=g.id,
einleitung=g.einleitung,
hauptteil=g.hauptteil,
fazit=g.fazit,
staerken=g.staerken,
schwaechen=g.schwaechen,
generated=g.generated,
edited=g.edited,
created_at=g.created_at
)
def _to_examiner_response(e: ExaminerResult) -> ExaminerResultResponse:
"""Konvertiert ExaminerResult zu Response."""
return ExaminerResultResponse(
examiner_id=e.examiner_id,
examiner_name=e.examiner_name,
grade_points=e.grade_points,
comment=e.comment,
submitted_at=e.submitted_at
)
def _to_student_klausur_response(sk: StudentKlausur) -> StudentKlausurResponse:
"""Konvertiert StudentKlausur zu Response."""
return StudentKlausurResponse(
id=sk.id,
student_name=sk.student_name,
student_id=sk.student_id,
file_path=sk.file_path,
file_name=sk.file_name,
ocr_text=sk.ocr_text,
status=sk.status,
criteria_scores={
k: {"score": v.score, "weight": v.weight, "annotations": v.annotations,
"comment": v.comment, "ai_suggestions": v.ai_suggestions}
for k, v in sk.criteria_scores.items()
},
annotations=[_to_annotation_response(a) for a in sk.annotations],
gutachten=_to_gutachten_response(sk.gutachten) if sk.gutachten else None,
raw_points=sk.raw_points,
grade_points=sk.grade_points,
grade_label=get_grade_label(sk.grade_points),
first_examiner=_to_examiner_response(sk.first_examiner) if sk.first_examiner else None,
second_examiner=_to_examiner_response(sk.second_examiner) if sk.second_examiner else None,
created_at=sk.created_at,
updated_at=sk.updated_at
)
def _to_klausur_response(k: AbiturKlausur) -> KlausurResponse:
"""Konvertiert AbiturKlausur zu Response."""
completed = [s for s in k.students if s.status == StudentKlausurStatus.COMPLETED]
avg_grade = None
if completed:
avg_grade = sum(s.grade_points for s in completed) / len(completed)
return KlausurResponse(
id=k.id,
title=k.title,
subject=k.subject,
modus=k.modus,
year=k.year,
semester=k.semester,
kurs=k.kurs,
class_id=k.class_id,
status=k.status,
text_sources=[_to_text_source_response(ts) for ts in k.text_sources],
erwartungshorizont=_to_erwartungshorizont_response(k.erwartungshorizont) if k.erwartungshorizont else None,
student_count=len(k.students),
completed_count=len(completed),
average_grade=avg_grade,
created_at=k.created_at,
updated_at=k.updated_at
)
# ============================================================================
# Background Tasks
# ============================================================================
async def _process_ocr(klausur_id: str, student_id: str, file_path: str):
"""Background Task für OCR-Verarbeitung."""
klausur = _klausuren.get(klausur_id)
if not klausur:
return
student = next((s for s in klausur.students if s.id == student_id), None)
if not student:
return
try:
student.status = StudentKlausurStatus.OCR_PROCESSING
student.updated_at = datetime.utcnow()
# OCR durchführen
processor = FileProcessor()
result = processor.process_file(file_path)
if result.success and result.text:
student.ocr_text = result.text
student.status = StudentKlausurStatus.OCR_COMPLETE
logger.info(f"OCR completed for {student_id}: {len(result.text)} characters")
else:
# Bei OCR-Fehlschlag auf UPLOADED zurücksetzen, damit manuell weitergearbeitet werden kann
student.ocr_text = None
student.status = StudentKlausurStatus.UPLOADED
logger.warning(f"OCR failed for {student_id}, keeping status as UPLOADED")
student.updated_at = datetime.utcnow()
except Exception as e:
logger.error(f"OCR error for {student_id}: {e}")
# Bei Fehlern auf UPLOADED zurücksetzen, nicht ERROR
student.status = StudentKlausurStatus.UPLOADED
student.ocr_text = None
student.updated_at = datetime.utcnow()
# ============================================================================
# API Endpoints - Klausuren
# ============================================================================
@router.post("/klausuren", response_model=KlausurResponse)
async def create_klausur(data: KlausurCreate):
"""Erstellt eine neue Klausur."""
klausur_id = str(uuid.uuid4())
now = datetime.utcnow()
klausur = AbiturKlausur(
id=klausur_id,
title=data.title,
subject=data.subject,
modus=data.modus,
year=data.year,
semester=data.semester,
kurs=data.kurs,
class_id=data.class_id,
status=KlausurStatus.DRAFT,
text_sources=[],
erwartungshorizont=None,
students=[],
created_at=now,
updated_at=now
)
_klausuren[klausur_id] = klausur
logger.info(f"Created klausur {klausur_id}: {data.title}")
return _to_klausur_response(klausur)
@router.get("/klausuren", response_model=List[KlausurResponse])
async def list_klausuren(
modus: Optional[KlausurModus] = None,
subject: Optional[str] = None,
year: Optional[int] = None,
status: Optional[KlausurStatus] = None
):
"""Listet alle Klausuren auf."""
klausuren = list(_klausuren.values())
if modus:
klausuren = [k for k in klausuren if k.modus == modus]
if subject:
klausuren = [k for k in klausuren if k.subject == subject]
if year:
klausuren = [k for k in klausuren if k.year == year]
if status:
klausuren = [k for k in klausuren if k.status == status]
klausuren.sort(key=lambda x: x.created_at, reverse=True)
return [_to_klausur_response(k) for k in klausuren]
@router.get("/klausuren/{klausur_id}", response_model=KlausurResponse)
async def get_klausur(klausur_id: str):
"""Ruft eine Klausur ab."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
return _to_klausur_response(klausur)
@router.put("/klausuren/{klausur_id}", response_model=KlausurResponse)
async def update_klausur(klausur_id: str, data: KlausurUpdate):
"""Aktualisiert eine Klausur."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
if data.title is not None:
klausur.title = data.title
if data.subject is not None:
klausur.subject = data.subject
if data.kurs is not None:
klausur.kurs = data.kurs
if data.status is not None:
klausur.status = data.status
klausur.updated_at = datetime.utcnow()
return _to_klausur_response(klausur)
@router.delete("/klausuren/{klausur_id}")
async def delete_klausur(klausur_id: str):
"""Löscht eine Klausur."""
if klausur_id not in _klausuren:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
klausur = _klausuren[klausur_id]
# Lösche hochgeladene Dateien
for student in klausur.students:
if student.file_path and os.path.exists(student.file_path):
try:
os.remove(student.file_path)
except Exception as e:
logger.warning(f"Could not delete file {student.file_path}: {e}")
del _klausuren[klausur_id]
logger.info(f"Deleted klausur {klausur_id}")
return {"status": "deleted", "id": klausur_id}
# ============================================================================
# API Endpoints - Text-Quellen
# ============================================================================
@router.post("/klausuren/{klausur_id}/text-sources", response_model=TextSourceResponse)
async def add_text_source(klausur_id: str, data: TextSourceCreate):
"""Fügt eine Textquelle zur Klausur hinzu."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
ts_id = str(uuid.uuid4())
now = datetime.utcnow()
# Status basierend auf Typ setzen
if data.source_type == TextSourceType.NIBIS:
license_status = TextSourceStatus.VERIFIED
elif data.source_type == TextSourceType.EIGENTEXT:
license_status = TextSourceStatus.VERIFIED
else:
license_status = TextSourceStatus.PENDING
text_source = TextSource(
id=ts_id,
source_type=data.source_type,
title=data.title,
author=data.author,
content=data.content,
nibis_id=data.nibis_id,
license_status=license_status,
license_info=data.license_info,
created_at=now
)
klausur.text_sources.append(text_source)
klausur.status = KlausurStatus.TEXT_SOURCE
klausur.updated_at = now
logger.info(f"Added text source {ts_id} to klausur {klausur_id}")
return _to_text_source_response(text_source)
@router.post("/text-sources/{text_source_id}/verify", response_model=TextSourceResponse)
async def verify_text_source(text_source_id: str, license_info: Dict[str, Any] = None):
"""Verifiziert eine Textquelle (Rights-Gate)."""
# Finde Textquelle
for klausur in _klausuren.values():
for ts in klausur.text_sources:
if ts.id == text_source_id:
ts.license_status = TextSourceStatus.VERIFIED
ts.license_info = license_info or {"verified_at": datetime.utcnow().isoformat()}
# Prüfe ob alle Textquellen verifiziert
all_verified = all(
t.license_status == TextSourceStatus.VERIFIED
for t in klausur.text_sources
)
if all_verified and klausur.text_sources:
klausur.status = KlausurStatus.ERWARTUNGSHORIZONT
klausur.updated_at = datetime.utcnow()
return _to_text_source_response(ts)
raise HTTPException(status_code=404, detail="Textquelle nicht gefunden")
@router.delete("/klausuren/{klausur_id}/text-sources/{text_source_id}")
async def delete_text_source(klausur_id: str, text_source_id: str):
"""Löscht eine Textquelle."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
klausur.text_sources = [ts for ts in klausur.text_sources if ts.id != text_source_id]
klausur.updated_at = datetime.utcnow()
return {"status": "deleted", "id": text_source_id}
# ============================================================================
# API Endpoints - NiBiS-Katalog
# ============================================================================
@router.get("/nibis/aufgaben", response_model=List[NiBiSAufgabe])
async def list_nibis_aufgaben(
subject: Optional[str] = None,
year: Optional[int] = None
):
"""Listet NiBiS-Aufgaben aus dem Katalog auf."""
aufgaben = list(_nibis_katalog.values())
if subject:
aufgaben = [a for a in aufgaben if a.subject.lower() == subject.lower()]
if year:
aufgaben = [a for a in aufgaben if a.year == year]
aufgaben.sort(key=lambda x: (x.year, x.subject), reverse=True)
return aufgaben
@router.get("/nibis/aufgaben/{aufgabe_id}", response_model=NiBiSAufgabe)
async def get_nibis_aufgabe(aufgabe_id: str):
"""Ruft eine NiBiS-Aufgabe ab."""
aufgabe = _nibis_katalog.get(aufgabe_id)
if not aufgabe:
raise HTTPException(status_code=404, detail="NiBiS-Aufgabe nicht gefunden")
return aufgabe
# ============================================================================
# API Endpoints - Erwartungshorizont
# ============================================================================
@router.post("/klausuren/{klausur_id}/erwartungshorizont/generate", response_model=ErwartungshorizontResponse)
async def generate_erwartungshorizont(klausur_id: str):
"""Generiert KI-basierten Erwartungshorizont."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
if not klausur.text_sources:
raise HTTPException(status_code=400, detail="Keine Textquellen vorhanden")
# Kombiniere alle Texte
combined_text = "\n\n".join(ts.content for ts in klausur.text_sources)
# Generiere Demo-Erwartungshorizont (in Produktion: KI-generiert)
eh_id = str(uuid.uuid4())
now = datetime.utcnow()
aufgaben = [
Aufgabe(
id=str(uuid.uuid4()),
nummer="1",
text="Analysieren Sie die sprachlichen und stilistischen Mittel des Textes.",
operator="analysieren",
anforderungsbereich=2,
erwartete_leistungen=[
"Erkennung von Metaphern und deren Wirkung",
"Analyse der Syntax und Satzstruktur",
"Einordnung in den historischen Kontext"
],
punkte=30
),
Aufgabe(
id=str(uuid.uuid4()),
nummer="2",
text="Erörtern Sie die Aktualität des Themas in Bezug auf die Gegenwart.",
operator="erörtern",
anforderungsbereich=3,
erwartete_leistungen=[
"Herausarbeitung der zentralen Thesen",
"Vergleich mit aktuellen gesellschaftlichen Entwicklungen",
"Differenzierte Stellungnahme mit Begründung"
],
punkte=40
),
Aufgabe(
id=str(uuid.uuid4()),
nummer="3",
text="Verfassen Sie einen Kommentar zum Thema aus persönlicher Perspektive.",
operator="gestalten",
anforderungsbereich=3,
erwartete_leistungen=[
"Einhaltung der Textsortenmerkmale (Kommentar)",
"Klare Positionierung und Argumentation",
"Sprachliche Angemessenheit und Stilsicherheit"
],
punkte=30
)
]
erwartungshorizont = Erwartungshorizont(
id=eh_id,
aufgaben=aufgaben,
max_points=100,
hinweise="Der Erwartungshorizont dient als Orientierung. Abweichende, aber schlüssige Lösungen sind möglich.",
generated=True,
created_at=now
)
klausur.erwartungshorizont = erwartungshorizont
klausur.status = KlausurStatus.COLLECTING
klausur.updated_at = now
logger.info(f"Generated erwartungshorizont for klausur {klausur_id}")
return _to_erwartungshorizont_response(erwartungshorizont)
@router.put("/klausuren/{klausur_id}/erwartungshorizont", response_model=ErwartungshorizontResponse)
async def update_erwartungshorizont(klausur_id: str, data: ErwartungshorizontCreate):
"""Aktualisiert den Erwartungshorizont."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
now = datetime.utcnow()
aufgaben = [
Aufgabe(
id=str(uuid.uuid4()),
nummer=a.nummer,
text=a.text,
operator=a.operator,
anforderungsbereich=a.anforderungsbereich,
erwartete_leistungen=a.erwartete_leistungen,
punkte=a.punkte
)
for a in data.aufgaben
]
if klausur.erwartungshorizont:
klausur.erwartungshorizont.aufgaben = aufgaben
klausur.erwartungshorizont.max_points = data.max_points
klausur.erwartungshorizont.hinweise = data.hinweise
else:
klausur.erwartungshorizont = Erwartungshorizont(
id=str(uuid.uuid4()),
aufgaben=aufgaben,
max_points=data.max_points,
hinweise=data.hinweise,
generated=False,
created_at=now
)
klausur.updated_at = now
return _to_erwartungshorizont_response(klausur.erwartungshorizont)
@router.get("/operators", response_model=Dict[str, List[Dict[str, str]]])
async def get_operators():
"""Gibt alle Operatoren nach Anforderungsbereich zurück."""
return OPERATORS
# ============================================================================
# API Endpoints - Schülerarbeiten
# ============================================================================
@router.post("/klausuren/{klausur_id}/students", response_model=StudentKlausurResponse)
async def add_student(
klausur_id: str,
background_tasks: BackgroundTasks,
student_name: str = Form(...),
student_id: Optional[str] = Form(None),
file: UploadFile = File(...)
):
"""Lädt eine Schülerarbeit hoch."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
# Validiere Dateiformat
allowed_extensions = {".pdf", ".png", ".jpg", ".jpeg"}
file_ext = Path(file.filename).suffix.lower() if file.filename else ""
if file_ext not in allowed_extensions:
raise HTTPException(
status_code=400,
detail=f"Ungültiges Dateiformat. Erlaubt: {', '.join(allowed_extensions)}"
)
sk_id = str(uuid.uuid4())
now = datetime.utcnow()
# Speichere Datei
file_path = UPLOAD_DIR / f"{sk_id}{file_ext}"
try:
content = await file.read()
with open(file_path, "wb") as f:
f.write(content)
except Exception as e:
logger.error(f"Upload error: {e}")
raise HTTPException(status_code=500, detail=f"Upload fehlgeschlagen: {str(e)}")
# Initialisiere Kriterien-Scores
criteria_scores = {
k: CriterionScore(
score=0,
weight=v["weight"],
annotations=[],
comment="",
ai_suggestions=[]
)
for k, v in DEFAULT_CRITERIA.items()
}
student_klausur = StudentKlausur(
id=sk_id,
student_name=student_name,
student_id=student_id,
file_path=str(file_path),
file_name=file.filename,
ocr_text=None,
status=StudentKlausurStatus.UPLOADED,
criteria_scores=criteria_scores,
annotations=[],
gutachten=None,
raw_points=0,
grade_points=0,
first_examiner=None,
second_examiner=None,
created_at=now,
updated_at=now
)
klausur.students.append(student_klausur)
klausur.status = KlausurStatus.CORRECTING
klausur.updated_at = now
# Starte OCR im Hintergrund
background_tasks.add_task(_process_ocr, klausur_id, sk_id, str(file_path))
logger.info(f"Added student {student_name} to klausur {klausur_id}")
return _to_student_klausur_response(student_klausur)
@router.get("/klausuren/{klausur_id}/students", response_model=List[StudentKlausurResponse])
async def list_students(klausur_id: str):
"""Listet alle Schülerarbeiten einer Klausur auf."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
return [_to_student_klausur_response(s) for s in klausur.students]
@router.get("/students/{student_id}", response_model=StudentKlausurResponse)
async def get_student(student_id: str):
"""Ruft eine Schülerarbeit ab."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
return _to_student_klausur_response(student)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.delete("/students/{student_id}")
async def delete_student(student_id: str):
"""Löscht eine Schülerarbeit und deren Datei."""
for klausur in _klausuren.values():
for i, student in enumerate(klausur.students):
if student.id == student_id:
# Lösche die Datei, falls vorhanden
if student.file_path:
file_path = Path(student.file_path)
if file_path.exists():
try:
file_path.unlink()
logger.info(f"Deleted file: {file_path}")
except Exception as e:
logger.warning(f"Could not delete file {file_path}: {e}")
# Entferne aus der Liste
klausur.students.pop(i)
klausur.updated_at = datetime.utcnow()
logger.info(f"Deleted student work {student_id} ({student.student_name})")
return {"success": True, "message": f"Schülerarbeit '{student.student_name}' gelöscht"}
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.get("/students/{student_id}/file")
async def get_student_file(student_id: str):
"""Liefert die Datei einer Schülerarbeit aus."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
if not student.file_path:
raise HTTPException(status_code=404, detail="Keine Datei vorhanden")
file_path = Path(student.file_path)
if not file_path.exists():
raise HTTPException(status_code=404, detail="Datei nicht gefunden")
# Determine media type
suffix = file_path.suffix.lower()
media_types = {
".pdf": "application/pdf",
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
}
media_type = media_types.get(suffix, "application/octet-stream")
return FileResponse(
path=str(file_path),
media_type=media_type,
filename=student.file_name or file_path.name
)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.post("/students/{student_id}/ocr")
async def start_ocr(student_id: str, background_tasks: BackgroundTasks):
"""Startet OCR für eine Schülerarbeit (erneut)."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
if not student.file_path:
raise HTTPException(status_code=400, detail="Keine Datei vorhanden")
student.status = StudentKlausurStatus.UPLOADED
student.ocr_text = None
student.updated_at = datetime.utcnow()
background_tasks.add_task(_process_ocr, klausur.id, student_id, student.file_path)
return {"status": "started", "student_id": student_id}
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
# ============================================================================
# API Endpoints - Bewertung
# ============================================================================
@router.post("/students/{student_id}/evaluate", response_model=StudentKlausurResponse)
async def evaluate_student(student_id: str):
"""Führt KI-basierte Bewertung durch."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
if student.status not in [StudentKlausurStatus.OCR_COMPLETE, StudentKlausurStatus.ANALYZING]:
raise HTTPException(
status_code=400,
detail=f"OCR muss abgeschlossen sein. Aktueller Status: {student.status}"
)
student.status = StudentKlausurStatus.ANALYZING
# Demo-Bewertung (in Produktion: KI-basiert)
import random
for criterion in student.criteria_scores:
base_score = random.randint(50, 90)
student.criteria_scores[criterion].score = base_score
student.criteria_scores[criterion].ai_suggestions = [
f"Verbesserungsvorschlag für {criterion}"
]
# Berechne Punkte
max_points = klausur.erwartungshorizont.max_points if klausur.erwartungshorizont else 100
student.raw_points = calculate_raw_points(student.criteria_scores, max_points)
percentage = (student.raw_points / max_points * 100) if max_points > 0 else 0
student.grade_points = calculate_15_point_grade(percentage)
student.status = StudentKlausurStatus.FIRST_EXAMINER
student.updated_at = datetime.utcnow()
logger.info(f"Evaluated student {student_id}: {student.grade_points} points")
return _to_student_klausur_response(student)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.put("/students/{student_id}/criteria", response_model=StudentKlausurResponse)
async def update_criteria(student_id: str, updates: Dict[str, CriterionScoreUpdate]):
"""Aktualisiert Kriterien-Bewertungen."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
for criterion, update in updates.items():
if criterion in student.criteria_scores:
student.criteria_scores[criterion].score = update.score
student.criteria_scores[criterion].annotations = update.annotations
student.criteria_scores[criterion].comment = update.comment
# Neuberechnung
max_points = klausur.erwartungshorizont.max_points if klausur.erwartungshorizont else 100
student.raw_points = calculate_raw_points(student.criteria_scores, max_points)
percentage = (student.raw_points / max_points * 100) if max_points > 0 else 0
student.grade_points = calculate_15_point_grade(percentage)
student.updated_at = datetime.utcnow()
return _to_student_klausur_response(student)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.post("/students/{student_id}/annotations", response_model=AnnotationResponse)
async def add_annotation(student_id: str, data: AnnotationCreate):
"""Fügt eine Annotation hinzu."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
annotation = Annotation(
id=str(uuid.uuid4()),
page=data.page,
x=data.x,
y=data.y,
width=data.width,
height=data.height,
text=data.text,
type=data.type,
color=data.color,
created_at=datetime.utcnow()
)
student.annotations.append(annotation)
student.updated_at = datetime.utcnow()
return _to_annotation_response(annotation)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.delete("/students/{student_id}/annotations/{annotation_id}")
async def delete_annotation(student_id: str, annotation_id: str):
"""Löscht eine Annotation."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
student.annotations = [a for a in student.annotations if a.id != annotation_id]
student.updated_at = datetime.utcnow()
return {"status": "deleted", "id": annotation_id}
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
# ============================================================================
# API Endpoints - Gutachten
# ============================================================================
@router.post("/students/{student_id}/gutachten/generate", response_model=GutachtenResponse)
async def generate_gutachten(student_id: str):
"""Generiert KI-basiertes Gutachten."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
now = datetime.utcnow()
grade_label = get_grade_label(student.grade_points)
# Demo-Gutachten (in Produktion: KI-generiert)
gutachten = Gutachten(
id=str(uuid.uuid4()),
einleitung=f"Die vorliegende Klausur von {student.student_name} im Fach {klausur.subject} "
f"wird im Folgenden nach den Kriterien des niedersächsischen Kerncurriculums bewertet.",
hauptteil=f"Die Arbeit zeigt in der inhaltlichen Auseinandersetzung mit dem Thema "
f"{'überzeugende' if student.grade_points >= 10 else 'grundlegende'} Kompetenzen. "
f"Die sprachliche Gestaltung ist {'angemessen' if student.grade_points >= 8 else 'ausbaufähig'}. "
f"Der Aufbau der Argumentation folgt {'einer klaren' if student.grade_points >= 10 else 'einer erkennbaren'} Struktur.",
fazit=f"Insgesamt erreicht die Arbeit {student.raw_points} von "
f"{klausur.erwartungshorizont.max_points if klausur.erwartungshorizont else 100} Punkten, "
f"was der Note {grade_label} entspricht.",
staerken=["Textverständnis", "Argumentationsaufbau"] if student.grade_points >= 10 else ["Grundlegende Texterfassung"],
schwaechen=["Sprachliche Vielfalt"] if student.grade_points >= 10 else ["Inhaltliche Tiefe", "Sprachliche Richtigkeit"],
generated=True,
edited=False,
created_at=now
)
student.gutachten = gutachten
student.updated_at = now
logger.info(f"Generated gutachten for student {student_id}")
return _to_gutachten_response(gutachten)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.put("/students/{student_id}/gutachten", response_model=GutachtenResponse)
async def update_gutachten(student_id: str, data: GutachtenCreate):
"""Aktualisiert das Gutachten."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
now = datetime.utcnow()
if student.gutachten:
student.gutachten.einleitung = data.einleitung
student.gutachten.hauptteil = data.hauptteil
student.gutachten.fazit = data.fazit
student.gutachten.staerken = data.staerken
student.gutachten.schwaechen = data.schwaechen
student.gutachten.edited = True
else:
student.gutachten = Gutachten(
id=str(uuid.uuid4()),
einleitung=data.einleitung,
hauptteil=data.hauptteil,
fazit=data.fazit,
staerken=data.staerken,
schwaechen=data.schwaechen,
generated=False,
edited=True,
created_at=now
)
student.updated_at = now
return _to_gutachten_response(student.gutachten)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
# ============================================================================
# API Endpoints - Note & Finalisierung
# ============================================================================
@router.post("/students/{student_id}/grade", response_model=StudentKlausurResponse)
async def calculate_grade(student_id: str):
"""Berechnet die Note neu."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
max_points = klausur.erwartungshorizont.max_points if klausur.erwartungshorizont else 100
student.raw_points = calculate_raw_points(student.criteria_scores, max_points)
percentage = (student.raw_points / max_points * 100) if max_points > 0 else 0
student.grade_points = calculate_15_point_grade(percentage)
student.updated_at = datetime.utcnow()
return _to_student_klausur_response(student)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.put("/students/{student_id}/finalize", response_model=StudentKlausurResponse)
async def finalize_student(student_id: str):
"""Schließt eine Schülerarbeit ab."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
student.status = StudentKlausurStatus.COMPLETED
student.updated_at = datetime.utcnow()
# Prüfe ob alle Schüler abgeschlossen sind
all_completed = all(
s.status == StudentKlausurStatus.COMPLETED
for s in klausur.students
)
if all_completed and klausur.students:
klausur.status = KlausurStatus.COMPLETED
klausur.updated_at = datetime.utcnow()
logger.info(f"Finalized student {student_id}")
return _to_student_klausur_response(student)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
# ============================================================================
# API Endpoints - Erst-/Zweitprüfer
# ============================================================================
@router.post("/students/{student_id}/examiner", response_model=StudentKlausurResponse)
async def submit_examiner_result(student_id: str, data: ExaminerResultCreate):
"""Reicht Prüfer-Ergebnis ein."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
now = datetime.utcnow()
result = ExaminerResult(
examiner_id=data.examiner_id,
examiner_name=data.examiner_name,
grade_points=data.grade_points,
comment=data.comment,
submitted_at=now
)
if student.status == StudentKlausurStatus.FIRST_EXAMINER:
student.first_examiner = result
student.status = StudentKlausurStatus.SECOND_EXAMINER
elif student.status == StudentKlausurStatus.SECOND_EXAMINER:
student.second_examiner = result
# Berechne Durchschnitt wenn beide Prüfer bewertet haben
if student.first_examiner:
avg = (student.first_examiner.grade_points + result.grade_points) / 2
student.grade_points = round(avg)
student.status = StudentKlausurStatus.COMPLETED
else:
raise HTTPException(
status_code=400,
detail=f"Kann Prüfer-Ergebnis nicht einreichen. Status: {student.status}"
)
student.updated_at = now
return _to_student_klausur_response(student)
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
@router.get("/examiner/queue", response_model=List[StudentKlausurResponse])
async def get_examiner_queue(examiner_role: str = "first"):
"""Gibt Warteschlange für Prüfer zurück."""
target_status = (
StudentKlausurStatus.FIRST_EXAMINER
if examiner_role == "first"
else StudentKlausurStatus.SECOND_EXAMINER
)
queue = []
for klausur in _klausuren.values():
for student in klausur.students:
if student.status == target_status:
queue.append(_to_student_klausur_response(student))
return queue
# ============================================================================
# API Endpoints - Fairness & Export
# ============================================================================
@router.get("/klausuren/{klausur_id}/fairness", response_model=FairnessReport)
async def get_fairness_report(klausur_id: str):
"""Erstellt Fairness-Bericht für Klausur."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
students = klausur.students
if not students:
raise HTTPException(status_code=400, detail="Keine Schülerarbeiten vorhanden")
# Notenverteilung
grade_distribution = {}
for s in students:
grade_distribution[s.grade_points] = grade_distribution.get(s.grade_points, 0) + 1
# Durchschnitte
avg_grade = sum(s.grade_points for s in students) / len(students)
criteria_averages = {}
for criterion in DEFAULT_CRITERIA:
scores = [s.criteria_scores[criterion].score for s in students if criterion in s.criteria_scores]
if scores:
criteria_averages[criterion] = sum(scores) / len(scores)
# Ausreißer (>2 Standardabweichungen)
grades = [s.grade_points for s in students]
mean = sum(grades) / len(grades)
variance = sum((g - mean) ** 2 for g in grades) / len(grades)
std_dev = variance ** 0.5
outliers = []
for s in students:
if abs(s.grade_points - mean) > 2 * std_dev:
outliers.append({
"student_name": s.student_name,
"grade_points": s.grade_points,
"deviation": s.grade_points - mean
})
# Empfehlungen
recommendations = []
if std_dev > 4:
recommendations.append("Hohe Streuung der Noten - Erwartungshorizont prüfen")
if avg_grade < 5:
recommendations.append("Durchschnitt unter 5 Punkten - Aufgabenstellung überprüfen")
if avg_grade > 12:
recommendations.append("Durchschnitt über 12 Punkten - Bewertungsmaßstab prüfen")
if outliers:
recommendations.append(f"{len(outliers)} Ausreißer gefunden - Einzelfälle prüfen")
return FairnessReport(
klausur_id=klausur_id,
total_students=len(students),
average_grade=avg_grade,
grade_distribution=grade_distribution,
criteria_averages=criteria_averages,
outliers=outliers,
recommendations=recommendations
)
@router.post("/klausuren/{klausur_id}/export")
async def export_klausur(
klausur_id: str,
student_ids: Optional[List[str]] = None,
include_gutachten: bool = True
):
"""Exportiert Klausur als PDF."""
klausur = _klausuren.get(klausur_id)
if not klausur:
raise HTTPException(status_code=404, detail="Klausur nicht gefunden")
# Filter Schüler falls IDs angegeben
students = klausur.students
if student_ids:
students = [s for s in students if s.id in student_ids]
if not students:
raise HTTPException(status_code=400, detail="Keine Schüler zum Exportieren")
# Demo-Response (in Produktion: PDF generieren)
return {
"status": "success",
"klausur_id": klausur_id,
"exported_students": len(students),
"message": "PDF-Export wird generiert..."
}
@router.get("/students/{student_id}/export-pdf")
async def export_student_pdf(student_id: str):
"""Exportiert einzelne Schülerarbeit als PDF."""
for klausur in _klausuren.values():
for student in klausur.students:
if student.id == student_id:
try:
pdf_service = PDFService()
# Erstelle HTML für Gutachten
grade_label = get_grade_label(student.grade_points)
html_content = f"""
{klausur.title}
Schüler/in: {student.student_name}
Fach: {klausur.subject}
Kurs: {klausur.kurs}
Datum: {student.created_at.strftime('%d.%m.%Y')}
Ergebnis
{student.grade_points} Punkte - {grade_label}
Erreichte Rohpunkte: {student.raw_points} / {klausur.erwartungshorizont.max_points if klausur.erwartungshorizont else 100}
Kriterien-Bewertung
| Kriterium |
Gewichtung |
Erreicht |
"""
for crit_name, crit_data in student.criteria_scores.items():
label = DEFAULT_CRITERIA.get(crit_name, {}).get("label", crit_name)
html_content += f"""
| {label} |
{int(crit_data.weight * 100)}% |
{crit_data.score}% |
"""
html_content += "
"
if student.gutachten:
html_content += f"""
Gutachten
{student.gutachten.einleitung}
{student.gutachten.hauptteil}
{student.gutachten.fazit}
Stärken
{''.join(f'- {s}
' for s in student.gutachten.staerken)}
Verbesserungsbereiche
{''.join(f'- {s}
' for s in student.gutachten.schwaechen)}
"""
html_content += """
Datum: _________________
Unterschrift Lehrkraft: _________________
"""
pdf_bytes = pdf_service.html_to_pdf(html_content)
return Response(
content=pdf_bytes,
media_type="application/pdf",
headers={
"Content-Disposition": f'attachment; filename="klausur_{student.student_name}_{klausur.subject}.pdf"'
}
)
except Exception as e:
logger.error(f"PDF export error: {e}")
raise HTTPException(status_code=500, detail=f"PDF-Export fehlgeschlagen: {str(e)}")
raise HTTPException(status_code=404, detail="Schülerarbeit nicht gefunden")
# ============================================================================
# API Endpoints - Hilfsfunktionen
# ============================================================================
@router.get("/grade-scale", response_model=Dict[str, Any])
async def get_grade_scale():
"""Gibt den 15-Punkte-Notenschlüssel zurück."""
return {
"thresholds": GRADE_THRESHOLDS,
"labels": GRADE_LABELS
}
@router.get("/criteria", response_model=Dict[str, Dict[str, Any]])
async def get_default_criteria():
"""Gibt die Standard-Bewertungskriterien zurück."""
return DEFAULT_CRITERIA
# ============================================================================
# Backwards-compatibility aliases (used by tests)
# ============================================================================
klausuren_db = _klausuren