grades_server.py

#!/usr/bin/env python3
"""
MCP Server für Notenverwaltung

Bietet Tools zum Zugriff auf Notenmeldungen aus PDF-Dateien:
- list_courses: Alle verfügbaren Kurse auflisten
- get_grades: Noten eines bestimmten Kurses abrufen
- get_student_transcript: Alle Noten eines Studierenden
- calculate_gpa: Durchschnittsnote berechnen
"""

import asyncio
import json
import os
from pathlib import Path
from typing import Any
import PyPDF2
import re

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# PDF-Verzeichnis
PDF_DIR = Path(__file__).parent.parent.parent / "data" / "pdfs"

app = Server("grades-server")


def extract_grades_from_pdf(pdf_path: str) -> dict:
    """Extrahiert Notendaten aus einem PDF."""

    with open(pdf_path, 'rb') as file:
        reader = PyPDF2.PdfReader(file)
        text = ""
        for page in reader.pages:
            text += page.extract_text()

    # Titel extrahieren
    title_match = re.search(r'(.+?)\s*-\s*WS\s*\d{4}/\d{2}', text)
    title = title_match.group(0) if title_match else "Unbekannter Kurs"

    # Professor extrahieren
    prof_match = re.search(r'Dozent:\s*(.+)', text)
    professor = prof_match.group(1).strip() if prof_match else "Unbekannt"

    # Noten extrahieren (Matrikelnummer Name Note)
    grades = []
    pattern = r'(\d{5})\s+([A-Za-zäöüÄÖÜß\s]+?)\s+(\d\.\d)'
    matches = re.finditer(pattern, text)

    for match in matches:
        matrikel = match.group(1)
        name = match.group(2).strip()
        grade = match.group(3)
        grades.append({
            "matrikel": matrikel,
            "name": name,
            "grade": float(grade)
        })

    return {
        "title": title,
        "professor": professor,
        "students": grades
    }


@app.list_tools()
async def list_tools() -> list[Tool]:
    """Liste aller verfügbaren Tools."""
    return [
        Tool(
            name="list_courses",
            description="Listet alle verfügbaren Kurse mit Notenmeldungen auf",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="get_grades",
            description="""Ruft alle Noten eines bestimmten Kurses ab.

Rückgabe-Format (strukturiertes JSON):
{
  "course_id": "string",
  "title": "string",
  "professor": "string",
  "students": [
    {
      "matrikel": "string",
      "name": "string",
      "grade": number
    }
  ],
  "statistics": {
    "average": number,
    "median": number,
    "count": number,
    "passed": number,
    "failed": number
  }
}""",
            inputSchema={
                "type": "object",
                "properties": {
                    "course_id": {
                        "type": "string",
                        "description": "ID des Kurses (z.B. 'datenbanken_ws2024')"
                    }
                },
                "required": ["course_id"]
            }
        ),
        Tool(
            name="get_student_transcript",
            description="Ruft alle Noten eines Studierenden über alle Kurse ab",
            inputSchema={
                "type": "object",
                "properties": {
                    "matrikel": {
                        "type": "string",
                        "description": "Matrikelnummer des Studierenden"
                    }
                },
                "required": ["matrikel"]
            }
        ),
        Tool(
            name="calculate_gpa",
            description="Berechnet die Durchschnittsnote eines Studierenden",
            inputSchema={
                "type": "object",
                "properties": {
                    "matrikel": {
                        "type": "string",
                        "description": "Matrikelnummer des Studierenden"
                    }
                },
                "required": ["matrikel"]
            }
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Tool-Aufruf verarbeiten."""

    if name == "list_courses":
        # Alle PDF-Dateien im Verzeichnis finden
        courses = []
        for pdf_file in PDF_DIR.glob("*.pdf"):
            course_id = pdf_file.stem
            data = extract_grades_from_pdf(str(pdf_file))
            courses.append({
                "id": course_id,
                "title": data["title"],
                "professor": data["professor"],
                "student_count": len(data["students"])
            })

        result = "Verfügbare Kurse:\n\n"
        for course in courses:
            result += f"• {course['title']}\n"
            result += f"  ID: {course['id']}\n"
            result += f"  Dozent: {course['professor']}\n"
            result += f"  Studierende: {course['student_count']}\n\n"

        return [TextContent(type="text", text=result)]

    elif name == "get_grades":
        course_id = arguments["course_id"]
        pdf_path = PDF_DIR / f"{course_id}.pdf"

        if not pdf_path.exists():
            return [TextContent(type="text", text=json.dumps({
                "error": f"Kurs '{course_id}' nicht gefunden."
            }))]

        data = extract_grades_from_pdf(str(pdf_path))

        # Statistiken berechnen
        grades = [s['grade'] for s in data['students']]
        grades_sorted = sorted(grades)
        count = len(grades)
        average = sum(grades) / count if count > 0 else 0
        median = grades_sorted[count // 2] if count > 0 else 0
        passed = sum(1 for g in grades if g <= 4.0)  # Deutsche Notenskala: <= 4.0 ist bestanden
        failed = count - passed

        # Strukturiertes JSON-Objekt erstellen
        result_data = {
            "course_id": course_id,
            "title": data['title'],
            "professor": data['professor'],
            "students": data['students'],
            "statistics": {
                "average": round(average, 2),
                "median": round(median, 2),
                "count": count,
                "passed": passed,
                "failed": failed
            }
        }

        # JSON zurückgeben (formatiert für bessere Lesbarkeit)
        return [TextContent(type="text", text=json.dumps(result_data, indent=2, ensure_ascii=False))]

    elif name == "get_student_transcript":
        matrikel = arguments["matrikel"]

        transcript = []
        for pdf_file in PDF_DIR.glob("*.pdf"):
            data = extract_grades_from_pdf(str(pdf_file))
            for student in data['students']:
                if student['matrikel'] == matrikel:
                    transcript.append({
                        "course": data['title'],
                        "grade": student['grade'],
                        "name": student['name']
                    })
                    break

        if not transcript:
            return [TextContent(type="text", text=f"Keine Noten für Matrikelnummer {matrikel} gefunden.")]

        student_name = transcript[0]['name']
        result = f"Notenspiegel für {student_name} (Matrikel: {matrikel})\n\n"
        result += "-" * 60 + "\n"
        result += f"{'Kurs':<45} {'Note':<6}\n"
        result += "-" * 60 + "\n"

        for entry in transcript:
            result += f"{entry['course']:<45} {entry['grade']:<6.1f}\n"

        avg = sum(e['grade'] for e in transcript) / len(transcript)
        result += "-" * 60 + "\n"
        result += f"Gesamtdurchschnitt: {avg:.2f}\n"

        return [TextContent(type="text", text=result)]

    elif name == "calculate_gpa":
        matrikel = arguments["matrikel"]

        grades = []
        student_name = None

        for pdf_file in PDF_DIR.glob("*.pdf"):
            data = extract_grades_from_pdf(str(pdf_file))
            for student in data['students']:
                if student['matrikel'] == matrikel:
                    grades.append(student['grade'])
                    student_name = student['name']
                    break

        if not grades:
            return [TextContent(type="text", text=f"Keine Noten für Matrikelnummer {matrikel} gefunden.")]

        gpa = sum(grades) / len(grades)
        result = f"Durchschnittsnote für {student_name} (Matrikel: {matrikel}):\n"
        result += f"GPA: {gpa:.2f}\n"
        result += f"Anzahl Kurse: {len(grades)}\n"

        return [TextContent(type="text", text=result)]

    raise ValueError(f"Unbekanntes Tool: {name}")


async def main():
    """Server starten."""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())


if __name__ == "__main__":
    asyncio.run(main())