""" 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)