bfdaf63ba9
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>
637 lines
21 KiB
Python
637 lines
21 KiB
Python
"""
|
|
Certificates API - Zeugnisverwaltung für BreakPilot.
|
|
|
|
Bietet Endpoints für:
|
|
- Erstellen und Verwalten von Zeugnissen
|
|
- PDF-Export von Zeugnissen
|
|
- Notenübersicht und Statistiken
|
|
- Archivierung in DSMS
|
|
|
|
Arbeitet zusammen mit:
|
|
- services/pdf_service.py für PDF-Generierung
|
|
- Gradebook für Notenverwaltung
|
|
"""
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
from enum import Enum
|
|
|
|
from fastapi import APIRouter, HTTPException, Response, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
# PDF service requires WeasyPrint with system libraries - make optional for CI
|
|
try:
|
|
from services.pdf_service import generate_certificate_pdf, SchoolInfo
|
|
_pdf_available = True
|
|
except (ImportError, OSError):
|
|
generate_certificate_pdf = None # type: ignore
|
|
SchoolInfo = None # type: ignore
|
|
_pdf_available = False
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/certificates", tags=["certificates"])
|
|
|
|
|
|
# =============================================================================
|
|
# Enums
|
|
# =============================================================================
|
|
|
|
class CertificateType(str, Enum):
|
|
"""Typen von Zeugnissen."""
|
|
HALBJAHR = "halbjahr" # Halbjahreszeugnis
|
|
JAHRES = "jahres" # Jahreszeugnis
|
|
ABSCHLUSS = "abschluss" # Abschlusszeugnis
|
|
ABGANG = "abgang" # Abgangszeugnis
|
|
UEBERGANG = "uebergang" # Übergangszeugnis
|
|
|
|
|
|
class CertificateStatus(str, Enum):
|
|
"""Status eines Zeugnisses."""
|
|
DRAFT = "draft" # Entwurf - noch in Bearbeitung
|
|
REVIEW = "review" # Zur Prüfung
|
|
APPROVED = "approved" # Genehmigt
|
|
ISSUED = "issued" # Ausgestellt
|
|
ARCHIVED = "archived" # Archiviert
|
|
|
|
|
|
class GradeType(str, Enum):
|
|
"""Notentyp."""
|
|
NUMERIC = "numeric" # 1-6
|
|
POINTS = "points" # 0-15 (Oberstufe)
|
|
TEXT = "text" # Verbal (Grundschule)
|
|
|
|
|
|
class BehaviorGrade(str, Enum):
|
|
"""Verhaltens-/Arbeitsnoten."""
|
|
A = "A" # Sehr gut
|
|
B = "B" # Gut
|
|
C = "C" # Befriedigend
|
|
D = "D" # Verbesserungswürdig
|
|
|
|
|
|
# =============================================================================
|
|
# Pydantic Models
|
|
# =============================================================================
|
|
|
|
class SchoolInfoModel(BaseModel):
|
|
"""Schulinformationen für Zeugnis."""
|
|
name: str
|
|
address: str
|
|
phone: str
|
|
email: str
|
|
website: Optional[str] = None
|
|
principal: Optional[str] = None
|
|
logo_path: Optional[str] = None
|
|
|
|
|
|
class SubjectGrade(BaseModel):
|
|
"""Note für ein Fach."""
|
|
name: str = Field(..., description="Fachname")
|
|
grade: str = Field(..., description="Note (1-6 oder A-D)")
|
|
points: Optional[int] = Field(None, description="Punkte (Oberstufe, 0-15)")
|
|
note: Optional[str] = Field(None, description="Bemerkung zum Fach")
|
|
|
|
|
|
class AttendanceInfo(BaseModel):
|
|
"""Anwesenheitsinformationen."""
|
|
days_absent: int = Field(0, description="Fehlende Tage gesamt")
|
|
days_excused: int = Field(0, description="Entschuldigte Tage")
|
|
days_unexcused: int = Field(0, description="Unentschuldigte Tage")
|
|
hours_absent: Optional[int] = Field(None, description="Fehlstunden gesamt")
|
|
|
|
|
|
class CertificateCreateRequest(BaseModel):
|
|
"""Request zum Erstellen eines neuen Zeugnisses."""
|
|
student_id: str = Field(..., description="ID des Schülers")
|
|
student_name: str = Field(..., description="Name des Schülers")
|
|
student_birthdate: str = Field(..., description="Geburtsdatum")
|
|
student_class: str = Field(..., description="Klasse")
|
|
school_year: str = Field(..., description="Schuljahr (z.B. '2024/2025')")
|
|
certificate_type: CertificateType = Field(..., description="Art des Zeugnisses")
|
|
subjects: List[SubjectGrade] = Field(..., description="Fachnoten")
|
|
attendance: AttendanceInfo = Field(default_factory=AttendanceInfo, description="Anwesenheit")
|
|
remarks: Optional[str] = Field(None, description="Bemerkungen")
|
|
class_teacher: str = Field(..., description="Klassenlehrer/in")
|
|
principal: str = Field(..., description="Schulleiter/in")
|
|
school_info: Optional[SchoolInfoModel] = Field(None, description="Schulinformationen")
|
|
issue_date: Optional[str] = Field(None, description="Ausstellungsdatum")
|
|
social_behavior: Optional[BehaviorGrade] = Field(None, description="Sozialverhalten")
|
|
work_behavior: Optional[BehaviorGrade] = Field(None, description="Arbeitsverhalten")
|
|
|
|
|
|
class CertificateUpdateRequest(BaseModel):
|
|
"""Request zum Aktualisieren eines Zeugnisses."""
|
|
subjects: Optional[List[SubjectGrade]] = None
|
|
attendance: Optional[AttendanceInfo] = None
|
|
remarks: Optional[str] = None
|
|
class_teacher: Optional[str] = None
|
|
principal: Optional[str] = None
|
|
social_behavior: Optional[BehaviorGrade] = None
|
|
work_behavior: Optional[BehaviorGrade] = None
|
|
status: Optional[CertificateStatus] = None
|
|
|
|
|
|
class CertificateResponse(BaseModel):
|
|
"""Response mit Zeugnisdaten."""
|
|
id: str
|
|
student_id: str
|
|
student_name: str
|
|
student_birthdate: str
|
|
student_class: str
|
|
school_year: str
|
|
certificate_type: CertificateType
|
|
subjects: List[SubjectGrade]
|
|
attendance: AttendanceInfo
|
|
remarks: Optional[str]
|
|
class_teacher: str
|
|
principal: str
|
|
school_info: Optional[SchoolInfoModel]
|
|
issue_date: Optional[str]
|
|
social_behavior: Optional[BehaviorGrade]
|
|
work_behavior: Optional[BehaviorGrade]
|
|
status: CertificateStatus
|
|
average_grade: Optional[float]
|
|
pdf_path: Optional[str]
|
|
dsms_cid: Optional[str]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class CertificateListResponse(BaseModel):
|
|
"""Response mit Liste von Zeugnissen."""
|
|
certificates: List[CertificateResponse]
|
|
total: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
class GradeStatistics(BaseModel):
|
|
"""Notenstatistiken für eine Klasse."""
|
|
class_name: str
|
|
school_year: str
|
|
certificate_type: CertificateType
|
|
student_count: int
|
|
average_grade: float
|
|
grade_distribution: Dict[str, int]
|
|
subject_averages: Dict[str, float]
|
|
|
|
|
|
# =============================================================================
|
|
# In-Memory Storage (Prototyp - später durch DB ersetzen)
|
|
# =============================================================================
|
|
|
|
_certificates_store: Dict[str, Dict[str, Any]] = {}
|
|
|
|
|
|
def _get_certificate(cert_id: str) -> Dict[str, Any]:
|
|
"""Holt Zeugnis aus dem Store."""
|
|
if cert_id not in _certificates_store:
|
|
raise HTTPException(status_code=404, detail=f"Zeugnis mit ID {cert_id} nicht gefunden")
|
|
return _certificates_store[cert_id]
|
|
|
|
|
|
def _save_certificate(cert_data: Dict[str, Any]) -> str:
|
|
"""Speichert Zeugnis und gibt ID zurück."""
|
|
cert_id = cert_data.get("id") or str(uuid.uuid4())
|
|
cert_data["id"] = cert_id
|
|
cert_data["updated_at"] = datetime.now()
|
|
if "created_at" not in cert_data:
|
|
cert_data["created_at"] = datetime.now()
|
|
_certificates_store[cert_id] = cert_data
|
|
return cert_id
|
|
|
|
|
|
def _calculate_average(subjects: List[Dict[str, Any]]) -> Optional[float]:
|
|
"""Berechnet Notendurchschnitt."""
|
|
numeric_grades = []
|
|
for subject in subjects:
|
|
grade = subject.get("grade", "")
|
|
try:
|
|
numeric = float(grade)
|
|
if 1 <= numeric <= 6:
|
|
numeric_grades.append(numeric)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
if numeric_grades:
|
|
return round(sum(numeric_grades) / len(numeric_grades), 2)
|
|
return None
|
|
|
|
|
|
# =============================================================================
|
|
# API Endpoints
|
|
# =============================================================================
|
|
|
|
@router.post("/", response_model=CertificateResponse)
|
|
async def create_certificate(request: CertificateCreateRequest):
|
|
"""
|
|
Erstellt ein neues Zeugnis.
|
|
|
|
Das Zeugnis wird als Entwurf gespeichert und kann später
|
|
bearbeitet, genehmigt und als PDF exportiert werden.
|
|
"""
|
|
logger.info(f"Creating new certificate for student: {request.student_name}")
|
|
|
|
subjects_list = [s.model_dump() for s in request.subjects]
|
|
|
|
cert_data = {
|
|
"student_id": request.student_id,
|
|
"student_name": request.student_name,
|
|
"student_birthdate": request.student_birthdate,
|
|
"student_class": request.student_class,
|
|
"school_year": request.school_year,
|
|
"certificate_type": request.certificate_type,
|
|
"subjects": subjects_list,
|
|
"attendance": request.attendance.model_dump(),
|
|
"remarks": request.remarks,
|
|
"class_teacher": request.class_teacher,
|
|
"principal": request.principal,
|
|
"school_info": request.school_info.model_dump() if request.school_info else None,
|
|
"issue_date": request.issue_date or datetime.now().strftime("%d.%m.%Y"),
|
|
"social_behavior": request.social_behavior,
|
|
"work_behavior": request.work_behavior,
|
|
"status": CertificateStatus.DRAFT,
|
|
"average_grade": _calculate_average(subjects_list),
|
|
"pdf_path": None,
|
|
"dsms_cid": None,
|
|
}
|
|
|
|
cert_id = _save_certificate(cert_data)
|
|
cert_data["id"] = cert_id
|
|
|
|
logger.info(f"Certificate created with ID: {cert_id}")
|
|
return CertificateResponse(**cert_data)
|
|
|
|
|
|
# IMPORTANT: Static routes must be defined BEFORE dynamic /{cert_id} route
|
|
# to prevent "types" or "behavior-grades" being matched as cert_id
|
|
|
|
@router.get("/types")
|
|
async def get_certificate_types():
|
|
"""
|
|
Gibt alle verfügbaren Zeugnistypen zurück.
|
|
"""
|
|
return {
|
|
"types": [
|
|
{"value": t.value, "label": _get_type_label(t)}
|
|
for t in CertificateType
|
|
]
|
|
}
|
|
|
|
|
|
@router.get("/behavior-grades")
|
|
async def get_behavior_grades():
|
|
"""
|
|
Gibt alle verfügbaren Verhaltensnoten zurück.
|
|
"""
|
|
labels = {
|
|
BehaviorGrade.A: "A - Sehr gut",
|
|
BehaviorGrade.B: "B - Gut",
|
|
BehaviorGrade.C: "C - Befriedigend",
|
|
BehaviorGrade.D: "D - Verbesserungswürdig"
|
|
}
|
|
return {
|
|
"grades": [
|
|
{"value": g.value, "label": labels[g]}
|
|
for g in BehaviorGrade
|
|
]
|
|
}
|
|
|
|
|
|
@router.get("/{cert_id}", response_model=CertificateResponse)
|
|
async def get_certificate(cert_id: str):
|
|
"""
|
|
Lädt ein gespeichertes Zeugnis.
|
|
"""
|
|
logger.info(f"Getting certificate: {cert_id}")
|
|
cert_data = _get_certificate(cert_id)
|
|
return CertificateResponse(**cert_data)
|
|
|
|
|
|
@router.get("/", response_model=CertificateListResponse)
|
|
async def list_certificates(
|
|
student_id: Optional[str] = Query(None, description="Filter nach Schüler-ID"),
|
|
class_name: Optional[str] = Query(None, description="Filter nach Klasse"),
|
|
school_year: Optional[str] = Query(None, description="Filter nach Schuljahr"),
|
|
certificate_type: Optional[CertificateType] = Query(None, description="Filter nach Zeugnistyp"),
|
|
status: Optional[CertificateStatus] = Query(None, description="Filter nach Status"),
|
|
page: int = Query(1, ge=1, description="Seitennummer"),
|
|
page_size: int = Query(20, ge=1, le=100, description="Einträge pro Seite")
|
|
):
|
|
"""
|
|
Listet alle gespeicherten Zeugnisse mit optionalen Filtern.
|
|
"""
|
|
logger.info("Listing certificates with filters")
|
|
|
|
# Filter anwenden
|
|
filtered_certs = list(_certificates_store.values())
|
|
|
|
if student_id:
|
|
filtered_certs = [c for c in filtered_certs if c.get("student_id") == student_id]
|
|
if class_name:
|
|
filtered_certs = [c for c in filtered_certs if c.get("student_class") == class_name]
|
|
if school_year:
|
|
filtered_certs = [c for c in filtered_certs if c.get("school_year") == school_year]
|
|
if certificate_type:
|
|
filtered_certs = [c for c in filtered_certs if c.get("certificate_type") == certificate_type]
|
|
if status:
|
|
filtered_certs = [c for c in filtered_certs if c.get("status") == status]
|
|
|
|
# Sortieren nach Erstelldatum (neueste zuerst)
|
|
filtered_certs.sort(key=lambda x: x.get("created_at", datetime.min), reverse=True)
|
|
|
|
# Paginierung
|
|
total = len(filtered_certs)
|
|
start = (page - 1) * page_size
|
|
end = start + page_size
|
|
paginated_certs = filtered_certs[start:end]
|
|
|
|
return CertificateListResponse(
|
|
certificates=[CertificateResponse(**c) for c in paginated_certs],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size
|
|
)
|
|
|
|
|
|
@router.put("/{cert_id}", response_model=CertificateResponse)
|
|
async def update_certificate(cert_id: str, request: CertificateUpdateRequest):
|
|
"""
|
|
Aktualisiert ein bestehendes Zeugnis.
|
|
"""
|
|
logger.info(f"Updating certificate: {cert_id}")
|
|
cert_data = _get_certificate(cert_id)
|
|
|
|
# Prüfen ob Zeugnis noch bearbeitbar ist
|
|
if cert_data.get("status") in [CertificateStatus.ISSUED, CertificateStatus.ARCHIVED]:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Zeugnis wurde bereits ausgestellt und kann nicht mehr bearbeitet werden"
|
|
)
|
|
|
|
# Nur übergebene Felder aktualisieren
|
|
update_data = request.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
if value is not None:
|
|
if key == "subjects":
|
|
cert_data[key] = [s if isinstance(s, dict) else s.model_dump() for s in value]
|
|
cert_data["average_grade"] = _calculate_average(cert_data["subjects"])
|
|
elif key == "attendance":
|
|
cert_data[key] = value if isinstance(value, dict) else value.model_dump()
|
|
else:
|
|
cert_data[key] = value
|
|
|
|
_save_certificate(cert_data)
|
|
|
|
return CertificateResponse(**cert_data)
|
|
|
|
|
|
@router.delete("/{cert_id}")
|
|
async def delete_certificate(cert_id: str):
|
|
"""
|
|
Löscht ein Zeugnis.
|
|
|
|
Nur Entwürfe können gelöscht werden.
|
|
"""
|
|
logger.info(f"Deleting certificate: {cert_id}")
|
|
cert_data = _get_certificate(cert_id)
|
|
|
|
if cert_data.get("status") != CertificateStatus.DRAFT:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Nur Zeugnis-Entwürfe können gelöscht werden"
|
|
)
|
|
|
|
del _certificates_store[cert_id]
|
|
return {"message": f"Zeugnis {cert_id} wurde gelöscht"}
|
|
|
|
|
|
@router.post("/{cert_id}/export-pdf")
|
|
async def export_certificate_pdf(cert_id: str):
|
|
"""
|
|
Exportiert ein Zeugnis als PDF.
|
|
"""
|
|
logger.info(f"Exporting certificate {cert_id} as PDF")
|
|
|
|
cert_data = _get_certificate(cert_id)
|
|
|
|
# PDF generieren
|
|
try:
|
|
pdf_bytes = generate_certificate_pdf(cert_data)
|
|
except Exception as e:
|
|
logger.error(f"Error generating PDF: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Fehler bei PDF-Generierung: {str(e)}")
|
|
|
|
# Dateiname erstellen (ASCII-safe für HTTP Header)
|
|
student_name = cert_data.get("student_name", "Zeugnis").replace(" ", "_")
|
|
school_year = cert_data.get("school_year", "").replace("/", "-")
|
|
cert_type = cert_data.get("certificate_type", "zeugnis")
|
|
filename = f"Zeugnis_{student_name}_{cert_type}_{school_year}.pdf"
|
|
|
|
# Für HTTP Header: ASCII-Fallback und UTF-8 encoded filename (RFC 5987)
|
|
from urllib.parse import quote
|
|
filename_ascii = filename.encode('ascii', 'replace').decode('ascii')
|
|
filename_encoded = quote(filename, safe='')
|
|
|
|
# PDF als Download zurückgeben
|
|
return Response(
|
|
content=pdf_bytes,
|
|
media_type="application/pdf",
|
|
headers={
|
|
"Content-Disposition": f"attachment; filename=\"{filename_ascii}\"; filename*=UTF-8''{filename_encoded}",
|
|
"Content-Length": str(len(pdf_bytes))
|
|
}
|
|
)
|
|
|
|
|
|
@router.post("/{cert_id}/submit-review")
|
|
async def submit_for_review(cert_id: str):
|
|
"""
|
|
Reicht Zeugnis zur Prüfung ein.
|
|
"""
|
|
logger.info(f"Submitting certificate {cert_id} for review")
|
|
cert_data = _get_certificate(cert_id)
|
|
|
|
if cert_data.get("status") != CertificateStatus.DRAFT:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Nur Entwürfe können zur Prüfung eingereicht werden"
|
|
)
|
|
|
|
# Prüfen ob alle Pflichtfelder ausgefüllt sind
|
|
if not cert_data.get("subjects"):
|
|
raise HTTPException(status_code=400, detail="Keine Fachnoten eingetragen")
|
|
|
|
cert_data["status"] = CertificateStatus.REVIEW
|
|
_save_certificate(cert_data)
|
|
|
|
return {"message": "Zeugnis wurde zur Prüfung eingereicht", "status": CertificateStatus.REVIEW}
|
|
|
|
|
|
@router.post("/{cert_id}/approve")
|
|
async def approve_certificate(cert_id: str):
|
|
"""
|
|
Genehmigt ein Zeugnis.
|
|
|
|
Erfordert Schulleiter-Rechte (in Produktion).
|
|
"""
|
|
logger.info(f"Approving certificate {cert_id}")
|
|
cert_data = _get_certificate(cert_id)
|
|
|
|
if cert_data.get("status") != CertificateStatus.REVIEW:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Nur Zeugnisse in Prüfung können genehmigt werden"
|
|
)
|
|
|
|
cert_data["status"] = CertificateStatus.APPROVED
|
|
_save_certificate(cert_data)
|
|
|
|
return {"message": "Zeugnis wurde genehmigt", "status": CertificateStatus.APPROVED}
|
|
|
|
|
|
@router.post("/{cert_id}/issue")
|
|
async def issue_certificate(cert_id: str):
|
|
"""
|
|
Stellt ein Zeugnis offiziell aus.
|
|
|
|
Nach Ausstellung kann das Zeugnis nicht mehr bearbeitet werden.
|
|
"""
|
|
logger.info(f"Issuing certificate {cert_id}")
|
|
cert_data = _get_certificate(cert_id)
|
|
|
|
if cert_data.get("status") != CertificateStatus.APPROVED:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Nur genehmigte Zeugnisse können ausgestellt werden"
|
|
)
|
|
|
|
cert_data["status"] = CertificateStatus.ISSUED
|
|
cert_data["issue_date"] = datetime.now().strftime("%d.%m.%Y")
|
|
_save_certificate(cert_data)
|
|
|
|
return {
|
|
"message": "Zeugnis wurde ausgestellt",
|
|
"status": CertificateStatus.ISSUED,
|
|
"issue_date": cert_data["issue_date"]
|
|
}
|
|
|
|
|
|
@router.get("/student/{student_id}", response_model=CertificateListResponse)
|
|
async def get_certificates_for_student(
|
|
student_id: str,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100)
|
|
):
|
|
"""
|
|
Lädt alle Zeugnisse für einen bestimmten Schüler.
|
|
"""
|
|
logger.info(f"Getting certificates for student: {student_id}")
|
|
|
|
filtered_certs = [
|
|
c for c in _certificates_store.values()
|
|
if c.get("student_id") == student_id
|
|
]
|
|
|
|
# Sortieren nach Schuljahr und Typ
|
|
filtered_certs.sort(key=lambda x: (x.get("school_year", ""), x.get("certificate_type", "")), reverse=True)
|
|
|
|
total = len(filtered_certs)
|
|
start = (page - 1) * page_size
|
|
end = start + page_size
|
|
paginated_certs = filtered_certs[start:end]
|
|
|
|
return CertificateListResponse(
|
|
certificates=[CertificateResponse(**c) for c in paginated_certs],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size
|
|
)
|
|
|
|
|
|
@router.get("/class/{class_name}/statistics", response_model=GradeStatistics)
|
|
async def get_class_statistics(
|
|
class_name: str,
|
|
school_year: str = Query(..., description="Schuljahr"),
|
|
certificate_type: CertificateType = Query(CertificateType.HALBJAHR, description="Zeugnistyp")
|
|
):
|
|
"""
|
|
Berechnet Notenstatistiken für eine Klasse.
|
|
"""
|
|
logger.info(f"Calculating statistics for class {class_name}")
|
|
|
|
# Filter Zeugnisse
|
|
class_certs = [
|
|
c for c in _certificates_store.values()
|
|
if c.get("student_class") == class_name
|
|
and c.get("school_year") == school_year
|
|
and c.get("certificate_type") == certificate_type
|
|
]
|
|
|
|
if not class_certs:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Keine Zeugnisse für Klasse {class_name} im Schuljahr {school_year} gefunden"
|
|
)
|
|
|
|
# Statistiken berechnen
|
|
all_grades = []
|
|
subject_grades: Dict[str, List[float]] = {}
|
|
grade_counts = {"1": 0, "2": 0, "3": 0, "4": 0, "5": 0, "6": 0}
|
|
|
|
for cert in class_certs:
|
|
avg = cert.get("average_grade")
|
|
if avg:
|
|
all_grades.append(avg)
|
|
# Runde für Verteilung
|
|
rounded = str(round(avg))
|
|
if rounded in grade_counts:
|
|
grade_counts[rounded] += 1
|
|
|
|
for subject in cert.get("subjects", []):
|
|
name = subject.get("name")
|
|
grade_str = subject.get("grade")
|
|
try:
|
|
grade = float(grade_str)
|
|
if name not in subject_grades:
|
|
subject_grades[name] = []
|
|
subject_grades[name].append(grade)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Fachdurchschnitte berechnen
|
|
subject_averages = {
|
|
name: round(sum(grades) / len(grades), 2)
|
|
for name, grades in subject_grades.items()
|
|
if grades
|
|
}
|
|
|
|
return GradeStatistics(
|
|
class_name=class_name,
|
|
school_year=school_year,
|
|
certificate_type=certificate_type,
|
|
student_count=len(class_certs),
|
|
average_grade=round(sum(all_grades) / len(all_grades), 2) if all_grades else 0.0,
|
|
grade_distribution=grade_counts,
|
|
subject_averages=subject_averages
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Helper Functions
|
|
# =============================================================================
|
|
|
|
def _get_type_label(cert_type: CertificateType) -> str:
|
|
"""Gibt menschenlesbare Labels für Zeugnistypen zurück."""
|
|
labels = {
|
|
CertificateType.HALBJAHR: "Halbjahreszeugnis",
|
|
CertificateType.JAHRES: "Jahreszeugnis",
|
|
CertificateType.ABSCHLUSS: "Abschlusszeugnis",
|
|
CertificateType.ABGANG: "Abgangszeugnis",
|
|
CertificateType.UEBERGANG: "Übergangszeugnis",
|
|
}
|
|
return labels.get(cert_type, cert_type.value)
|