v4.1.0: DMS email documents, category-specific Nachweis linking, version system
- Save cover email body as DMS document with new 'email' context type - Show email body separately from attachments in email detail view - Add per-category DMS document assignment in quarterly confirmation (Studiennachweis, Einkommenssituation, Vermögenssituation) - Add VERSION file and context processor for automatic version display - Add MCP server, agent system, import/export, and new migrations - Update compose files and production environment template Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
SysAdmin Agent
1
app/mcp_server/tools/__init__.py
Normal file
1
app/mcp_server/tools/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# MCP Tools
|
||||
59
app/mcp_server/tools/helpers.py
Normal file
59
app/mcp_server/tools/helpers.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""
|
||||
Hilfsfunktionen für MCP-Tool-Implementierungen.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import date, datetime, time
|
||||
from decimal import Decimal
|
||||
from uuid import UUID
|
||||
|
||||
|
||||
def serialize_value(value):
|
||||
"""Konvertiert Django-Feldwerte in JSON-serialisierbare Typen."""
|
||||
if isinstance(value, UUID):
|
||||
return str(value)
|
||||
if isinstance(value, Decimal):
|
||||
return float(value)
|
||||
if isinstance(value, (date, datetime)):
|
||||
return value.isoformat()
|
||||
if isinstance(value, time):
|
||||
return value.isoformat()
|
||||
return value
|
||||
|
||||
|
||||
def model_to_dict(instance, fields=None, exclude=None) -> dict:
|
||||
"""
|
||||
Konvertiert eine Django-Model-Instanz in ein serialisierbares Dict.
|
||||
|
||||
Args:
|
||||
instance: Django Model Instanz
|
||||
fields: Nur diese Felder einschließen (None = alle)
|
||||
exclude: Diese Felder ausschließen
|
||||
"""
|
||||
exclude = exclude or []
|
||||
result = {}
|
||||
for field in instance._meta.fields:
|
||||
name = field.name
|
||||
if fields and name not in fields:
|
||||
continue
|
||||
if name in exclude:
|
||||
continue
|
||||
value = getattr(instance, name)
|
||||
# ForeignKey: _id-Suffix-Wert (nicht ganzes Objekt)
|
||||
result[name] = serialize_value(value)
|
||||
|
||||
# Auch ForeignKey-IDs explizit aufnehmen (z.B. konto_id)
|
||||
for field in instance._meta.fields:
|
||||
if hasattr(field, "attname") and field.attname != field.name:
|
||||
attname = field.attname
|
||||
if attname not in result:
|
||||
result[attname] = serialize_value(getattr(instance, attname))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def format_result(data) -> str:
|
||||
"""Gibt Daten als formatiertes JSON zurück."""
|
||||
return json.dumps(data, ensure_ascii=False, indent=2, default=str)
|
||||
755
app/mcp_server/tools/lesen.py
Normal file
755
app/mcp_server/tools/lesen.py
Normal file
@@ -0,0 +1,755 @@
|
||||
"""
|
||||
Lese-Tools für den MCP Server der Stiftungsverwaltung.
|
||||
|
||||
Alle Tools:
|
||||
- Prüfen die Rolle (readonly/editor/admin erforderlich)
|
||||
- Wenden PII-Maskierung an (außer bei admin)
|
||||
- Schreiben Audit-Log-Einträge
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from django.db.models import Q, Sum
|
||||
|
||||
from mcp_server.audit import log_mcp_read
|
||||
from mcp_server.auth import require_role
|
||||
from mcp_server.privacy import apply_privacy_filter, apply_privacy_filter_list
|
||||
from mcp_server.tools.helpers import format_result, model_to_dict
|
||||
|
||||
|
||||
def _get_role() -> str:
|
||||
from mcp_server.auth import get_current_role, require_role as _require
|
||||
role = get_current_role()
|
||||
_require(role)
|
||||
return role
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Destinatäre
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def destinataer_suchen(
|
||||
suchbegriff: str = "",
|
||||
aktiv: bool | None = None,
|
||||
familienzweig: str = "",
|
||||
limit: int = 20,
|
||||
) -> str:
|
||||
"""
|
||||
Sucht Destinatäre nach Name, Familienzweig oder Aktivstatus.
|
||||
|
||||
Args:
|
||||
suchbegriff: Freitext (Vor-/Nachname, Institution)
|
||||
aktiv: True=nur Aktive, False=nur Inaktive, None=alle
|
||||
familienzweig: Filtert nach Familienzweig (hauptzweig/nebenzweig/verwandt/anderer)
|
||||
limit: Maximale Anzahl Ergebnisse (max. 100)
|
||||
"""
|
||||
from stiftung.models import Destinataer
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 100)
|
||||
|
||||
qs = Destinataer.objects.all()
|
||||
if suchbegriff:
|
||||
qs = qs.filter(
|
||||
Q(vorname__icontains=suchbegriff)
|
||||
| Q(nachname__icontains=suchbegriff)
|
||||
| Q(institution__icontains=suchbegriff)
|
||||
)
|
||||
if aktiv is not None:
|
||||
qs = qs.filter(aktiv=aktiv)
|
||||
if familienzweig:
|
||||
qs = qs.filter(familienzweig=familienzweig)
|
||||
|
||||
qs = qs.order_by("nachname", "vorname")[:limit]
|
||||
|
||||
# Reduzierte Felder für Listen-Ausgabe
|
||||
results = []
|
||||
for obj in qs:
|
||||
item = {
|
||||
"id": str(obj.id),
|
||||
"vorname": obj.vorname,
|
||||
"nachname": obj.nachname,
|
||||
"familienzweig": obj.familienzweig,
|
||||
"aktiv": obj.aktiv,
|
||||
"berufsgruppe": obj.berufsgruppe,
|
||||
"ort": obj.ort,
|
||||
"email": obj.email,
|
||||
}
|
||||
results.append(apply_privacy_filter(item, "destinataer", role))
|
||||
|
||||
log_mcp_read(role, "destinataer", "Destinatär-Suche", f"Suche: '{suchbegriff}', {len(results)} Ergebnisse")
|
||||
return format_result({"anzahl": len(results), "destinataere": results})
|
||||
|
||||
|
||||
def destinataer_details(destinataer_id: str) -> str:
|
||||
"""
|
||||
Gibt vollständige Details eines Destinatärs zurück.
|
||||
|
||||
Args:
|
||||
destinataer_id: UUID des Destinatärs
|
||||
"""
|
||||
from stiftung.models import Destinataer, DestinataerUnterstuetzung, Foerderung
|
||||
|
||||
role = _get_role()
|
||||
|
||||
try:
|
||||
obj = Destinataer.objects.get(id=destinataer_id)
|
||||
except Destinataer.DoesNotExist:
|
||||
return format_result({"fehler": f"Destinatär {destinataer_id} nicht gefunden"})
|
||||
|
||||
data = model_to_dict(obj)
|
||||
data = apply_privacy_filter(data, "destinataer", role)
|
||||
|
||||
# Aktuelle Unterstützungen
|
||||
unterstuetzungen = list(
|
||||
DestinataerUnterstuetzung.objects.filter(destinataer=obj)
|
||||
.exclude(status="storniert")
|
||||
.order_by("-faellig_am")[:10]
|
||||
.values("id", "betrag", "faellig_am", "status", "beschreibung")
|
||||
)
|
||||
for u in unterstuetzungen:
|
||||
for k, v in u.items():
|
||||
from mcp_server.tools.helpers import serialize_value
|
||||
u[k] = serialize_value(v)
|
||||
|
||||
# Förderungen
|
||||
foerderungen = list(
|
||||
Foerderung.objects.filter(destinataer=obj)
|
||||
.order_by("-jahr")[:10]
|
||||
.values("id", "jahr", "betrag", "kategorie", "status")
|
||||
)
|
||||
for f in foerderungen:
|
||||
for k, v in f.items():
|
||||
from mcp_server.tools.helpers import serialize_value
|
||||
f[k] = serialize_value(v)
|
||||
|
||||
data["aktuelle_unterstuetzungen"] = unterstuetzungen
|
||||
data["foerderungen"] = foerderungen
|
||||
|
||||
name = f"{obj.vorname} {obj.nachname}"
|
||||
log_mcp_read(role, "destinataer", name, f"Details abgerufen für {name}")
|
||||
return format_result(data)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Ländereien
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def land_suchen(
|
||||
suchbegriff: str = "",
|
||||
gemeinde: str = "",
|
||||
limit: int = 20,
|
||||
) -> str:
|
||||
"""
|
||||
Sucht Ländereien nach Bezeichnung, Gemarkung oder Gemeinde.
|
||||
|
||||
Args:
|
||||
suchbegriff: Freitext (Bezeichnung, Gemarkung)
|
||||
gemeinde: Filtert nach Gemeinde
|
||||
limit: Maximale Anzahl Ergebnisse (max. 100)
|
||||
"""
|
||||
from stiftung.models import Land
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 100)
|
||||
|
||||
qs = Land.objects.all()
|
||||
if suchbegriff:
|
||||
qs = qs.filter(
|
||||
Q(gemeinde__icontains=suchbegriff)
|
||||
| Q(gemarkung__icontains=suchbegriff)
|
||||
| Q(flur__icontains=suchbegriff)
|
||||
| Q(lfd_nr__icontains=suchbegriff)
|
||||
| Q(adresse__icontains=suchbegriff)
|
||||
)
|
||||
if gemeinde:
|
||||
qs = qs.filter(gemeinde__icontains=gemeinde)
|
||||
|
||||
qs = qs.order_by("gemeinde", "gemarkung")[:limit]
|
||||
|
||||
results = []
|
||||
for obj in qs:
|
||||
results.append({
|
||||
"id": str(obj.id),
|
||||
"bezeichnung": str(obj),
|
||||
"lfd_nr": obj.lfd_nr,
|
||||
"gemeinde": obj.gemeinde,
|
||||
"gemarkung": obj.gemarkung,
|
||||
"flur": obj.flur,
|
||||
"flurstueck": obj.flurstueck,
|
||||
"groesse_qm": float(obj.groesse_qm) if obj.groesse_qm else None,
|
||||
"aktiv_verpachtet": obj.neue_verpachtungen.filter(status="aktiv").exists(),
|
||||
})
|
||||
|
||||
log_mcp_read(role, "land", "Länderei-Suche", f"Suche: '{suchbegriff}', {len(results)} Ergebnisse")
|
||||
return format_result({"anzahl": len(results), "laendereien": results})
|
||||
|
||||
|
||||
def land_details(land_id: str) -> str:
|
||||
"""
|
||||
Gibt vollständige Details einer Länderei zurück.
|
||||
|
||||
Args:
|
||||
land_id: UUID der Länderei
|
||||
"""
|
||||
from stiftung.models import Land, LandVerpachtung
|
||||
|
||||
role = _get_role()
|
||||
|
||||
try:
|
||||
obj = Land.objects.get(id=land_id)
|
||||
except Land.DoesNotExist:
|
||||
return format_result({"fehler": f"Länderei {land_id} nicht gefunden"})
|
||||
|
||||
data = model_to_dict(obj)
|
||||
|
||||
# Aktive Verpachtungen
|
||||
verpachtungen = []
|
||||
for v in obj.neue_verpachtungen.all().order_by("-pachtbeginn")[:5]:
|
||||
verpachtungen.append({
|
||||
"id": str(v.id),
|
||||
"paechter": str(v.paechter) if v.paechter else None,
|
||||
"pachtbeginn": v.pachtbeginn.isoformat() if v.pachtbeginn else None,
|
||||
"pachtende": v.pachtende.isoformat() if v.pachtende else None,
|
||||
"pachtzins_pauschal": float(v.pachtzins_pauschal) if v.pachtzins_pauschal else None,
|
||||
"status": v.status,
|
||||
})
|
||||
|
||||
data["verpachtungen"] = verpachtungen
|
||||
log_mcp_read(role, "land", str(obj), f"Land-Details abgerufen: {obj}")
|
||||
return format_result(data)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Pächter
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def paechter_suchen(
|
||||
suchbegriff: str = "",
|
||||
limit: int = 20,
|
||||
) -> str:
|
||||
"""
|
||||
Sucht Pächter nach Name.
|
||||
|
||||
Args:
|
||||
suchbegriff: Freitext (Vor-/Nachname)
|
||||
limit: Maximale Anzahl Ergebnisse (max. 100)
|
||||
"""
|
||||
from stiftung.models import Paechter
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 100)
|
||||
|
||||
qs = Paechter.objects.all()
|
||||
if suchbegriff:
|
||||
qs = qs.filter(
|
||||
Q(vorname__icontains=suchbegriff) | Q(nachname__icontains=suchbegriff)
|
||||
)
|
||||
|
||||
qs = qs.order_by("nachname", "vorname")[:limit]
|
||||
|
||||
results = []
|
||||
for obj in qs:
|
||||
item = {
|
||||
"id": str(obj.id),
|
||||
"vorname": obj.vorname,
|
||||
"nachname": obj.nachname,
|
||||
"personentyp": obj.personentyp,
|
||||
"ort": obj.ort,
|
||||
"email": obj.email,
|
||||
"telefon": obj.telefon,
|
||||
"aktive_verpachtungen": obj.neue_verpachtungen.filter(status="aktiv").count() if hasattr(obj, "neue_verpachtungen") else 0,
|
||||
}
|
||||
results.append(apply_privacy_filter(item, "paechter", role))
|
||||
|
||||
log_mcp_read(role, "paechter", "Pächter-Suche", f"Suche: '{suchbegriff}', {len(results)} Ergebnisse")
|
||||
return format_result({"anzahl": len(results), "paechter": results})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Konten
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def konten_uebersicht() -> str:
|
||||
"""
|
||||
Gibt eine Übersicht aller Stiftungskonten mit aktuellem Saldo zurück.
|
||||
"""
|
||||
from stiftung.models import StiftungsKonto
|
||||
|
||||
role = _get_role()
|
||||
|
||||
konten = []
|
||||
gesamt_saldo = 0.0
|
||||
for konto in StiftungsKonto.objects.filter(aktiv=True).order_by("bank_name"):
|
||||
saldo = float(konto.saldo) if konto.saldo else 0.0
|
||||
gesamt_saldo += saldo
|
||||
konten.append({
|
||||
"id": str(konto.id),
|
||||
"kontoname": konto.kontoname,
|
||||
"bank_name": konto.bank_name,
|
||||
"konto_typ": konto.konto_typ,
|
||||
"saldo": saldo,
|
||||
"saldo_datum": konto.saldo_datum.isoformat() if konto.saldo_datum else None,
|
||||
"zinssatz": float(konto.zinssatz) if konto.zinssatz else None,
|
||||
# IBAN nur für Admin
|
||||
"iban": konto.iban if role == "admin" else "****" + konto.iban[-4:] if konto.iban and len(konto.iban) > 4 else "****",
|
||||
})
|
||||
|
||||
log_mcp_read(role, "stiftungskonto", "Kontenübersicht", f"{len(konten)} Konten abgerufen")
|
||||
return format_result({
|
||||
"konten": konten,
|
||||
"gesamt_saldo": round(gesamt_saldo, 2),
|
||||
"anzahl_konten": len(konten),
|
||||
})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Verwaltungskosten
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def verwaltungskosten(
|
||||
jahr: int | None = None,
|
||||
kategorie: str = "",
|
||||
status: str = "",
|
||||
limit: int = 50,
|
||||
) -> str:
|
||||
"""
|
||||
Listet Verwaltungskosten auf.
|
||||
|
||||
Args:
|
||||
jahr: Filtert nach Jahr (z.B. 2024)
|
||||
kategorie: Filtert nach Kategorie (rechnung_intern, bueroausstattung, ...)
|
||||
status: Filtert nach Status (geplant, bezahlt, ...)
|
||||
limit: Maximale Anzahl (max. 200)
|
||||
"""
|
||||
from stiftung.models import Verwaltungskosten
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 200)
|
||||
|
||||
qs = Verwaltungskosten.objects.all()
|
||||
if jahr:
|
||||
qs = qs.filter(datum__year=jahr)
|
||||
if kategorie:
|
||||
qs = qs.filter(kategorie=kategorie)
|
||||
if status:
|
||||
qs = qs.filter(status=status)
|
||||
|
||||
qs = qs.order_by("-datum")[:limit]
|
||||
|
||||
results = []
|
||||
gesamt = 0.0
|
||||
for obj in qs:
|
||||
betrag = float(obj.betrag) if obj.betrag else 0.0
|
||||
gesamt += betrag
|
||||
results.append({
|
||||
"id": str(obj.id),
|
||||
"bezeichnung": obj.bezeichnung,
|
||||
"kategorie": obj.kategorie,
|
||||
"betrag": betrag,
|
||||
"datum": obj.datum.isoformat(),
|
||||
"lieferant_firma": obj.lieferant_firma,
|
||||
"status": obj.status,
|
||||
"rechnungsnummer": obj.rechnungsnummer,
|
||||
})
|
||||
|
||||
log_mcp_read(role, "verwaltungskosten", "Verwaltungskosten", f"{len(results)} Einträge")
|
||||
return format_result({
|
||||
"anzahl": len(results),
|
||||
"gesamt_betrag": round(gesamt, 2),
|
||||
"verwaltungskosten": results,
|
||||
})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Transaktionen
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def transaktionen_suchen(
|
||||
suchbegriff: str = "",
|
||||
konto_id: str = "",
|
||||
von_datum: str = "",
|
||||
bis_datum: str = "",
|
||||
transaction_type: str = "",
|
||||
limit: int = 50,
|
||||
) -> str:
|
||||
"""
|
||||
Sucht Banktransaktionen.
|
||||
|
||||
Args:
|
||||
suchbegriff: Freitext in Verwendungszweck oder Empfänger
|
||||
konto_id: UUID des Kontos (optional)
|
||||
von_datum: Startdatum YYYY-MM-DD (optional)
|
||||
bis_datum: Enddatum YYYY-MM-DD (optional)
|
||||
transaction_type: eingang/ausgang/dauerauftrag/... (optional)
|
||||
limit: Maximale Anzahl (max. 200)
|
||||
"""
|
||||
from stiftung.models import BankTransaction
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 200)
|
||||
|
||||
qs = BankTransaction.objects.select_related("konto").all()
|
||||
if suchbegriff:
|
||||
qs = qs.filter(
|
||||
Q(verwendungszweck__icontains=suchbegriff)
|
||||
| Q(empfaenger_zahlungspflichtiger__icontains=suchbegriff)
|
||||
)
|
||||
if konto_id:
|
||||
qs = qs.filter(konto_id=konto_id)
|
||||
if von_datum:
|
||||
qs = qs.filter(datum__gte=von_datum)
|
||||
if bis_datum:
|
||||
qs = qs.filter(datum__lte=bis_datum)
|
||||
if transaction_type:
|
||||
qs = qs.filter(transaction_type=transaction_type)
|
||||
|
||||
qs = qs.order_by("-datum")[:limit]
|
||||
|
||||
results = []
|
||||
for obj in qs:
|
||||
results.append({
|
||||
"id": str(obj.id),
|
||||
"datum": obj.datum.isoformat(),
|
||||
"betrag": float(obj.betrag),
|
||||
"waehrung": obj.waehrung,
|
||||
"verwendungszweck": obj.verwendungszweck[:200],
|
||||
"empfaenger_zahlungspflichtiger": obj.empfaenger_zahlungspflichtiger,
|
||||
"transaction_type": obj.transaction_type,
|
||||
"status": obj.status,
|
||||
"konto": obj.konto.kontoname if obj.konto else None,
|
||||
})
|
||||
|
||||
log_mcp_read(role, "banktransaction", "Transaktionssuche", f"Suche: '{suchbegriff}', {len(results)} Ergebnisse")
|
||||
return format_result({"anzahl": len(results), "transaktionen": results})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Dokumente
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def dokument_suchen(
|
||||
suchbegriff: str = "",
|
||||
kontext: str = "",
|
||||
limit: int = 30,
|
||||
) -> str:
|
||||
"""
|
||||
Sucht Dokumente im DMS nach Titel, Beschreibung oder Kontext.
|
||||
|
||||
Args:
|
||||
suchbegriff: Freitext (Titel, Beschreibung, Volltext)
|
||||
kontext: Dokumententyp (pachtvertrag, antrag, rechnung, ...)
|
||||
limit: Maximale Anzahl (max. 100)
|
||||
"""
|
||||
from stiftung.models import DokumentDatei
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 100)
|
||||
|
||||
qs = DokumentDatei.objects.all()
|
||||
if suchbegriff:
|
||||
qs = qs.filter(
|
||||
Q(titel__icontains=suchbegriff)
|
||||
| Q(beschreibung__icontains=suchbegriff)
|
||||
| Q(inhaltstext__icontains=suchbegriff)
|
||||
)
|
||||
if kontext:
|
||||
qs = qs.filter(kontext=kontext)
|
||||
|
||||
qs = qs.order_by("-erstellt_am")[:limit]
|
||||
|
||||
results = []
|
||||
for obj in qs:
|
||||
results.append({
|
||||
"id": str(obj.id),
|
||||
"titel": obj.titel,
|
||||
"kontext": obj.kontext,
|
||||
"beschreibung": obj.beschreibung[:200] if obj.beschreibung else "",
|
||||
"dateityp": obj.dateityp,
|
||||
"dateigroesse": obj.dateigroesse,
|
||||
"dateiname_original": obj.dateiname_original,
|
||||
})
|
||||
|
||||
log_mcp_read(role, "dokumentlink", "Dokumentsuche", f"Suche: '{suchbegriff}', {len(results)} Ergebnisse")
|
||||
return format_result({"anzahl": len(results), "dokumente": results})
|
||||
|
||||
|
||||
def dokument_details(dokument_id: str) -> str:
|
||||
"""
|
||||
Gibt Details eines Dokuments zurück (ohne Dateiinhalt).
|
||||
|
||||
Args:
|
||||
dokument_id: UUID des Dokuments
|
||||
"""
|
||||
from stiftung.models import DokumentDatei
|
||||
|
||||
role = _get_role()
|
||||
|
||||
try:
|
||||
obj = DokumentDatei.objects.get(id=dokument_id)
|
||||
except DokumentDatei.DoesNotExist:
|
||||
return format_result({"fehler": f"Dokument {dokument_id} nicht gefunden"})
|
||||
|
||||
data = {
|
||||
"id": str(obj.id),
|
||||
"titel": obj.titel,
|
||||
"kontext": obj.kontext,
|
||||
"beschreibung": obj.beschreibung,
|
||||
"dateityp": obj.dateityp,
|
||||
"dateigroesse": obj.dateigroesse,
|
||||
"dateiname_original": obj.dateiname_original,
|
||||
# Verknüpfungen
|
||||
"land_id": str(obj.land_id) if obj.land_id else None,
|
||||
"paechter_id": str(obj.paechter_id) if obj.paechter_id else None,
|
||||
}
|
||||
# Inhaltstext nur für Nicht-Binary-Dokumente und wenn vorhanden
|
||||
if obj.inhaltstext:
|
||||
data["inhaltsvorschau"] = obj.inhaltstext[:500]
|
||||
|
||||
log_mcp_read(role, "dokumentlink", obj.titel, f"Dokumentdetails abgerufen: {obj.titel}")
|
||||
return format_result(data)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Termine
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def termine_anzeigen(
|
||||
von_datum: str = "",
|
||||
bis_datum: str = "",
|
||||
kategorie: str = "",
|
||||
prioritaet: str = "",
|
||||
limit: int = 50,
|
||||
) -> str:
|
||||
"""
|
||||
Zeigt Kalendereinträge und Termine an.
|
||||
|
||||
Args:
|
||||
von_datum: Startdatum YYYY-MM-DD (optional, Standard: heute)
|
||||
bis_datum: Enddatum YYYY-MM-DD (optional)
|
||||
kategorie: termin/zahlung/deadline/geburtstag/vertrag/pruefung/sonstiges
|
||||
prioritaet: niedrig/normal/hoch/kritisch
|
||||
limit: Maximale Anzahl (max. 200)
|
||||
"""
|
||||
from datetime import date as date_type
|
||||
from stiftung.models import StiftungsKalenderEintrag
|
||||
|
||||
role = _get_role()
|
||||
limit = min(limit, 200)
|
||||
|
||||
qs = StiftungsKalenderEintrag.objects.all()
|
||||
if von_datum:
|
||||
qs = qs.filter(datum__gte=von_datum)
|
||||
else:
|
||||
qs = qs.filter(datum__gte=date_type.today())
|
||||
if bis_datum:
|
||||
qs = qs.filter(datum__lte=bis_datum)
|
||||
if kategorie:
|
||||
qs = qs.filter(kategorie=kategorie)
|
||||
if prioritaet:
|
||||
qs = qs.filter(prioritaet=prioritaet)
|
||||
|
||||
qs = qs.order_by("datum", "uhrzeit")[:limit]
|
||||
|
||||
results = []
|
||||
for obj in qs:
|
||||
results.append({
|
||||
"id": str(obj.id),
|
||||
"titel": obj.titel,
|
||||
"datum": obj.datum.isoformat(),
|
||||
"uhrzeit": obj.uhrzeit.isoformat() if obj.uhrzeit else None,
|
||||
"ganztags": obj.ganztags,
|
||||
"kategorie": obj.kategorie,
|
||||
"prioritaet": obj.prioritaet,
|
||||
"beschreibung": obj.beschreibung[:300] if obj.beschreibung else "",
|
||||
"destinataer_id": str(obj.destinataer_id) if obj.destinataer_id else None,
|
||||
})
|
||||
|
||||
log_mcp_read(role, "system", "Terminübersicht", f"{len(results)} Termine abgerufen")
|
||||
return format_result({"anzahl": len(results), "termine": results})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Globale Suche & Dashboard
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def globale_suche(suchbegriff: str, limit_pro_typ: int = 5) -> str:
|
||||
"""
|
||||
Sucht über alle Entitätstypen gleichzeitig.
|
||||
|
||||
Args:
|
||||
suchbegriff: Suchbegriff (mindestens 2 Zeichen)
|
||||
limit_pro_typ: Ergebnisse pro Entitätstyp (max. 20)
|
||||
"""
|
||||
from stiftung.models import (
|
||||
BankTransaction, Destinataer, Land, Paechter,
|
||||
StiftungsKalenderEintrag, Verwaltungskosten,
|
||||
)
|
||||
|
||||
role = _get_role()
|
||||
|
||||
if len(suchbegriff) < 2:
|
||||
return format_result({"fehler": "Suchbegriff muss mindestens 2 Zeichen lang sein"})
|
||||
|
||||
limit_pro_typ = min(limit_pro_typ, 20)
|
||||
ergebnisse = {}
|
||||
|
||||
# Destinatäre
|
||||
dest = Destinataer.objects.filter(
|
||||
Q(vorname__icontains=suchbegriff) | Q(nachname__icontains=suchbegriff)
|
||||
)[:limit_pro_typ]
|
||||
ergebnisse["destinataere"] = [
|
||||
{"id": str(d.id), "name": f"{d.vorname} {d.nachname}", "typ": "destinataer"}
|
||||
for d in dest
|
||||
]
|
||||
|
||||
# Ländereien
|
||||
laender = Land.objects.filter(
|
||||
Q(bezeichnung__icontains=suchbegriff) | Q(gemarkung__icontains=suchbegriff)
|
||||
)[:limit_pro_typ]
|
||||
ergebnisse["laendereien"] = [
|
||||
{"id": str(l.id), "name": str(l), "typ": "land"}
|
||||
for l in laender
|
||||
]
|
||||
|
||||
# Pächter
|
||||
paechter = Paechter.objects.filter(
|
||||
Q(vorname__icontains=suchbegriff) | Q(nachname__icontains=suchbegriff)
|
||||
)[:limit_pro_typ]
|
||||
ergebnisse["paechter"] = [
|
||||
{"id": str(p.id), "name": f"{p.vorname} {p.nachname}", "typ": "paechter"}
|
||||
for p in paechter
|
||||
]
|
||||
|
||||
# Transaktionen
|
||||
transaktionen = BankTransaction.objects.filter(
|
||||
Q(verwendungszweck__icontains=suchbegriff)
|
||||
| Q(empfaenger_zahlungspflichtiger__icontains=suchbegriff)
|
||||
)[:limit_pro_typ]
|
||||
ergebnisse["transaktionen"] = [
|
||||
{"id": str(t.id), "verwendungszweck": t.verwendungszweck[:100], "betrag": float(t.betrag), "datum": t.datum.isoformat(), "typ": "transaktion"}
|
||||
for t in transaktionen
|
||||
]
|
||||
|
||||
log_mcp_read(role, "system", "Globale Suche", f"Suche: '{suchbegriff}'")
|
||||
return format_result(ergebnisse)
|
||||
|
||||
|
||||
def dashboard() -> str:
|
||||
"""
|
||||
Gibt eine Übersicht der wichtigsten Stiftungsdaten zurück.
|
||||
"""
|
||||
from datetime import date as date_type
|
||||
from stiftung.models import (
|
||||
BankTransaction, Destinataer, DestinataerUnterstuetzung,
|
||||
Land, LandVerpachtung, StiftungsKalenderEintrag, StiftungsKonto,
|
||||
)
|
||||
|
||||
role = _get_role()
|
||||
heute = date_type.today()
|
||||
|
||||
# Konten-Gesamtsaldo
|
||||
konten = StiftungsKonto.objects.filter(aktiv=True)
|
||||
gesamt_saldo = sum(float(k.saldo or 0) for k in konten)
|
||||
|
||||
# Destinatäre
|
||||
aktive_dest = Destinataer.objects.filter(aktiv=True).count()
|
||||
|
||||
# Offene Zahlungen
|
||||
offene_zahlungen = DestinataerUnterstuetzung.objects.filter(
|
||||
status__in=["geplant", "faellig", "nachweis_eingereicht", "freigegeben"]
|
||||
).aggregate(anzahl=Sum("betrag"))
|
||||
offene_zahlungen_betrag = float(offene_zahlungen["anzahl"] or 0)
|
||||
offene_zahlungen_anzahl = DestinataerUnterstuetzung.objects.filter(
|
||||
status__in=["geplant", "faellig", "nachweis_eingereicht", "freigegeben"]
|
||||
).count()
|
||||
|
||||
# Fällige Termine (nächste 30 Tage)
|
||||
from datetime import timedelta
|
||||
naechste_termine = StiftungsKalenderEintrag.objects.filter(
|
||||
datum__gte=heute,
|
||||
datum__lte=heute + timedelta(days=30),
|
||||
).order_by("datum")[:5]
|
||||
|
||||
termine_liste = [
|
||||
{"titel": t.titel, "datum": t.datum.isoformat(), "prioritaet": t.prioritaet, "kategorie": t.kategorie}
|
||||
for t in naechste_termine
|
||||
]
|
||||
|
||||
# Aktive Verpachtungen
|
||||
aktive_verpachtungen = LandVerpachtung.objects.filter(status="aktiv").count()
|
||||
|
||||
log_mcp_read(role, "system", "Dashboard", "Dashboard abgerufen")
|
||||
return format_result({
|
||||
"stand": heute.isoformat(),
|
||||
"finanzen": {
|
||||
"gesamt_saldo_eur": round(gesamt_saldo, 2),
|
||||
"anzahl_konten": konten.count(),
|
||||
},
|
||||
"destinataere": {
|
||||
"aktiv": aktive_dest,
|
||||
},
|
||||
"zahlungen": {
|
||||
"offen_anzahl": offene_zahlungen_anzahl,
|
||||
"offen_betrag_eur": round(offene_zahlungen_betrag, 2),
|
||||
},
|
||||
"verpachtungen": {
|
||||
"aktiv": aktive_verpachtungen,
|
||||
},
|
||||
"naechste_termine": termine_liste,
|
||||
})
|
||||
|
||||
|
||||
def statistiken() -> str:
|
||||
"""
|
||||
Gibt detaillierte Statistiken der Stiftungsverwaltung zurück.
|
||||
"""
|
||||
from datetime import date as date_type
|
||||
from stiftung.models import (
|
||||
BankTransaction, Destinataer, Foerderung,
|
||||
Land, LandVerpachtung, Verwaltungskosten,
|
||||
)
|
||||
|
||||
role = _get_role()
|
||||
aktuelles_jahr = date_type.today().year
|
||||
|
||||
# Förderungen dieses Jahr
|
||||
foerderungen_jahr = Foerderung.objects.filter(jahr=aktuelles_jahr)
|
||||
foerderungen_gesamt = foerderungen_jahr.aggregate(summe=Sum("betrag"))
|
||||
|
||||
# Destinatäre nach Familienzweig
|
||||
from django.db.models import Count
|
||||
dest_zweige = list(
|
||||
Destinataer.objects.filter(aktiv=True)
|
||||
.values("familienzweig")
|
||||
.annotate(anzahl=Count("id"))
|
||||
.order_by("-anzahl")
|
||||
)
|
||||
|
||||
# Verwaltungskosten dieses Jahr
|
||||
vk_jahr = Verwaltungskosten.objects.filter(datum__year=aktuelles_jahr)
|
||||
vk_gesamt = vk_jahr.aggregate(summe=Sum("betrag"))
|
||||
|
||||
# Ländereien
|
||||
laender_ges = Land.objects.count()
|
||||
laender_verpachtet = Land.objects.filter(
|
||||
neue_verpachtungen__status="aktiv"
|
||||
).distinct().count()
|
||||
|
||||
log_mcp_read(role, "system", "Statistiken", f"Statistiken für {aktuelles_jahr} abgerufen")
|
||||
return format_result({
|
||||
"jahr": aktuelles_jahr,
|
||||
"foerderungen": {
|
||||
"anzahl": foerderungen_jahr.count(),
|
||||
"gesamt_betrag_eur": float(foerderungen_gesamt["summe"] or 0),
|
||||
},
|
||||
"verwaltungskosten": {
|
||||
"anzahl": vk_jahr.count(),
|
||||
"gesamt_betrag_eur": float(vk_gesamt["summe"] or 0),
|
||||
},
|
||||
"destinataere_nach_zweig": dest_zweige,
|
||||
"laendereien": {
|
||||
"gesamt": laender_ges,
|
||||
"aktiv_verpachtet": laender_verpachtet,
|
||||
},
|
||||
})
|
||||
577
app/mcp_server/tools/schreiben.py
Normal file
577
app/mcp_server/tools/schreiben.py
Normal file
@@ -0,0 +1,577 @@
|
||||
"""
|
||||
Schreib-Tools für den MCP Server der Stiftungsverwaltung.
|
||||
|
||||
Alle Tools:
|
||||
- Prüfen die Rolle (editor oder admin erforderlich)
|
||||
- Schreiben Audit-Log-Einträge
|
||||
- Validieren Pflichtfelder vor dem Speichern
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from mcp_server.audit import log_mcp_create, log_mcp_update
|
||||
from mcp_server.auth import can_write, require_role
|
||||
from mcp_server.tools.helpers import format_result
|
||||
|
||||
|
||||
def _require_write_role() -> str:
|
||||
from mcp_server.auth import get_current_role
|
||||
|
||||
role = get_current_role()
|
||||
require_role(role)
|
||||
if not can_write(role):
|
||||
raise PermissionError(
|
||||
f"Rolle '{role}' hat keine Schreibrechte. "
|
||||
"editor- oder admin-Rolle erforderlich."
|
||||
)
|
||||
return role
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Destinatäre
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def destinataer_anlegen(
|
||||
vorname: str,
|
||||
nachname: str,
|
||||
familienzweig: str = "",
|
||||
email: str = "",
|
||||
telefon: str = "",
|
||||
geburtsdatum: str = "",
|
||||
ort: str = "",
|
||||
plz: str = "",
|
||||
strasse: str = "",
|
||||
berufsgruppe: str = "",
|
||||
notizen: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Legt einen neuen Destinatär an.
|
||||
|
||||
Args:
|
||||
vorname: Vorname (Pflichtfeld)
|
||||
nachname: Nachname (Pflichtfeld)
|
||||
familienzweig: hauptzweig/nebenzweig/verwandt/anderer
|
||||
email: E-Mail-Adresse
|
||||
telefon: Telefonnummer
|
||||
geburtsdatum: Geburtsdatum YYYY-MM-DD
|
||||
ort: Wohnort
|
||||
plz: Postleitzahl
|
||||
strasse: Straße und Hausnummer
|
||||
berufsgruppe: student/wissenschaftler/künstler/sozialarbeiter/umweltschützer/andere
|
||||
notizen: Freitext-Notizen
|
||||
"""
|
||||
from stiftung.models import Destinataer
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
kwargs = {
|
||||
"vorname": vorname.strip(),
|
||||
"nachname": nachname.strip(),
|
||||
"aktiv": True,
|
||||
}
|
||||
if familienzweig:
|
||||
kwargs["familienzweig"] = familienzweig
|
||||
if email:
|
||||
kwargs["email"] = email
|
||||
if telefon:
|
||||
kwargs["telefon"] = telefon
|
||||
if geburtsdatum:
|
||||
kwargs["geburtsdatum"] = geburtsdatum
|
||||
if ort:
|
||||
kwargs["ort"] = ort
|
||||
if plz:
|
||||
kwargs["plz"] = plz
|
||||
if strasse:
|
||||
kwargs["strasse"] = strasse
|
||||
if berufsgruppe:
|
||||
kwargs["berufsgruppe"] = berufsgruppe
|
||||
if notizen:
|
||||
kwargs["notizen"] = notizen
|
||||
|
||||
obj = Destinataer.objects.create(**kwargs)
|
||||
log_mcp_create(role, "destinataer", str(obj.id), f"{vorname} {nachname}")
|
||||
return format_result({"erfolg": True, "id": str(obj.id), "name": f"{vorname} {nachname}"})
|
||||
|
||||
|
||||
def destinataer_aktualisieren(
|
||||
destinataer_id: str,
|
||||
vorname: str = "",
|
||||
nachname: str = "",
|
||||
email: str = "",
|
||||
telefon: str = "",
|
||||
ort: str = "",
|
||||
plz: str = "",
|
||||
strasse: str = "",
|
||||
aktiv: bool | None = None,
|
||||
notizen: str = "",
|
||||
familienzweig: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Aktualisiert einen bestehenden Destinatär.
|
||||
|
||||
Args:
|
||||
destinataer_id: UUID des Destinatärs (Pflichtfeld)
|
||||
vorname: Neuer Vorname (optional)
|
||||
nachname: Neuer Nachname (optional)
|
||||
email: Neue E-Mail (optional)
|
||||
telefon: Neue Telefonnummer (optional)
|
||||
ort: Neuer Ort (optional)
|
||||
plz: Neue PLZ (optional)
|
||||
strasse: Neue Straße (optional)
|
||||
aktiv: Aktivstatus (optional)
|
||||
notizen: Neue Notizen (optional)
|
||||
familienzweig: Neuer Familienzweig (optional)
|
||||
"""
|
||||
from stiftung.models import Destinataer
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
try:
|
||||
obj = Destinataer.objects.get(id=destinataer_id)
|
||||
except Destinataer.DoesNotExist:
|
||||
return format_result({"fehler": f"Destinatär {destinataer_id} nicht gefunden"})
|
||||
|
||||
changes = {}
|
||||
update_fields = []
|
||||
|
||||
def _set(field, value):
|
||||
if value != "" and value is not None:
|
||||
old = getattr(obj, field)
|
||||
if str(old) != str(value):
|
||||
changes[field] = {"alt": str(old), "neu": str(value)}
|
||||
setattr(obj, field, value)
|
||||
update_fields.append(field)
|
||||
|
||||
_set("vorname", vorname)
|
||||
_set("nachname", nachname)
|
||||
_set("email", email)
|
||||
_set("telefon", telefon)
|
||||
_set("ort", ort)
|
||||
_set("plz", plz)
|
||||
_set("strasse", strasse)
|
||||
_set("notizen", notizen)
|
||||
_set("familienzweig", familienzweig)
|
||||
if aktiv is not None:
|
||||
_set("aktiv", aktiv)
|
||||
|
||||
if not update_fields:
|
||||
return format_result({"erfolg": True, "hinweis": "Keine Änderungen"})
|
||||
|
||||
obj.save(update_fields=update_fields)
|
||||
name = f"{obj.vorname} {obj.nachname}"
|
||||
log_mcp_update(role, "destinataer", str(obj.id), name, changes)
|
||||
return format_result({"erfolg": True, "id": str(obj.id), "geaenderte_felder": list(changes.keys())})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Förderungen & Unterstützungen
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def foerderung_anlegen(
|
||||
destinataer_id: str,
|
||||
jahr: int,
|
||||
betrag: float,
|
||||
kategorie: str = "anderes",
|
||||
bemerkungen: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Legt eine neue Förderung für einen Destinatär an.
|
||||
|
||||
Args:
|
||||
destinataer_id: UUID des Destinatärs (Pflichtfeld)
|
||||
jahr: Förderjahr (Pflichtfeld)
|
||||
betrag: Förderbetrag in EUR (Pflichtfeld)
|
||||
kategorie: bildung/forschung/kultur/soziales/umwelt/anderes
|
||||
bemerkungen: Freitext
|
||||
"""
|
||||
from datetime import date
|
||||
|
||||
from stiftung.models import Destinataer, Foerderung
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
try:
|
||||
destinataer = Destinataer.objects.get(id=destinataer_id)
|
||||
except Destinataer.DoesNotExist:
|
||||
return format_result({"fehler": f"Destinatär {destinataer_id} nicht gefunden"})
|
||||
|
||||
foerderung = Foerderung.objects.create(
|
||||
destinataer=destinataer,
|
||||
jahr=jahr,
|
||||
betrag=betrag,
|
||||
kategorie=kategorie,
|
||||
bemerkungen=bemerkungen,
|
||||
status="beantragt",
|
||||
antragsdatum=date.today(),
|
||||
)
|
||||
|
||||
name = f"{destinataer.vorname} {destinataer.nachname} – {jahr}"
|
||||
log_mcp_create(role, "foerderung", str(foerderung.id), name)
|
||||
return format_result({"erfolg": True, "id": str(foerderung.id), "foerderung": name})
|
||||
|
||||
|
||||
def unterstuetzung_anlegen(
|
||||
destinataer_id: str,
|
||||
konto_id: str,
|
||||
betrag: float,
|
||||
faellig_am: str,
|
||||
beschreibung: str = "",
|
||||
verwendungszweck: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Legt eine neue Unterstützungszahlung für einen Destinatär an.
|
||||
|
||||
Args:
|
||||
destinataer_id: UUID des Destinatärs (Pflichtfeld)
|
||||
konto_id: UUID des Zahlungskontos (Pflichtfeld)
|
||||
betrag: Betrag in EUR (Pflichtfeld)
|
||||
faellig_am: Fälligkeitsdatum YYYY-MM-DD (Pflichtfeld)
|
||||
beschreibung: Kurzbeschreibung (optional)
|
||||
verwendungszweck: Verwendungszweck für Überweisung (optional)
|
||||
"""
|
||||
from stiftung.models import Destinataer, DestinataerUnterstuetzung, StiftungsKonto
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
try:
|
||||
destinataer = Destinataer.objects.get(id=destinataer_id)
|
||||
except Destinataer.DoesNotExist:
|
||||
return format_result({"fehler": f"Destinatär {destinataer_id} nicht gefunden"})
|
||||
|
||||
try:
|
||||
konto = StiftungsKonto.objects.get(id=konto_id)
|
||||
except StiftungsKonto.DoesNotExist:
|
||||
return format_result({"fehler": f"Konto {konto_id} nicht gefunden"})
|
||||
|
||||
unterstuetzung = DestinataerUnterstuetzung.objects.create(
|
||||
destinataer=destinataer,
|
||||
konto=konto,
|
||||
betrag=betrag,
|
||||
faellig_am=faellig_am,
|
||||
beschreibung=beschreibung,
|
||||
verwendungszweck=verwendungszweck,
|
||||
status="geplant",
|
||||
)
|
||||
|
||||
name = f"{destinataer.vorname} {destinataer.nachname} – {faellig_am}"
|
||||
log_mcp_create(role, "destinataer", str(unterstuetzung.id), name)
|
||||
return format_result({"erfolg": True, "id": str(unterstuetzung.id)})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Ländereien & Verpachtungen
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def land_anlegen(
|
||||
lfd_nr: str,
|
||||
amtsgericht: str,
|
||||
gemeinde: str,
|
||||
gemarkung: str,
|
||||
flur: str,
|
||||
flurstueck: str,
|
||||
groesse_qm: float,
|
||||
verpachtete_gesamtflaeche: float = 0.0,
|
||||
adresse: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Legt eine neue Länderei an.
|
||||
|
||||
Args:
|
||||
lfd_nr: Laufende Nummer (Pflichtfeld, eindeutig)
|
||||
amtsgericht: Zuständiges Amtsgericht (Pflichtfeld)
|
||||
gemeinde: Gemeinde (Pflichtfeld)
|
||||
gemarkung: Gemarkung (Pflichtfeld)
|
||||
flur: Flur (Pflichtfeld)
|
||||
flurstueck: Flurstück (Pflichtfeld)
|
||||
groesse_qm: Gesamtgröße in Quadratmetern (Pflichtfeld)
|
||||
verpachtete_gesamtflaeche: Verpachtete Fläche in qm (Standard: 0)
|
||||
adresse: Adresse/Ortsangabe (optional)
|
||||
"""
|
||||
from stiftung.models import Land
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
if Land.objects.filter(lfd_nr=lfd_nr).exists():
|
||||
return format_result({"fehler": f"Länderei mit lfd_nr '{lfd_nr}' existiert bereits"})
|
||||
|
||||
land = Land.objects.create(
|
||||
lfd_nr=lfd_nr,
|
||||
amtsgericht=amtsgericht,
|
||||
gemeinde=gemeinde,
|
||||
gemarkung=gemarkung,
|
||||
flur=flur,
|
||||
flurstueck=flurstueck,
|
||||
groesse_qm=groesse_qm,
|
||||
verpachtete_gesamtflaeche=verpachtete_gesamtflaeche,
|
||||
adresse=adresse,
|
||||
)
|
||||
|
||||
log_mcp_create(role, "land", str(land.id), str(land))
|
||||
return format_result({"erfolg": True, "id": str(land.id), "bezeichnung": str(land)})
|
||||
|
||||
|
||||
def verpachtung_anlegen(
|
||||
land_id: str,
|
||||
paechter_id: str,
|
||||
vertragsnummer: str,
|
||||
pachtbeginn: str,
|
||||
verpachtete_flaeche: float,
|
||||
pachtzins_pauschal: float,
|
||||
zahlungsweise: str = "jaehrlich",
|
||||
pachtende: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Legt einen neuen Pachtvertrag für eine Länderei an.
|
||||
|
||||
Args:
|
||||
land_id: UUID der Länderei (Pflichtfeld)
|
||||
paechter_id: UUID des Pächters (Pflichtfeld)
|
||||
vertragsnummer: Eindeutige Vertragsnummer (Pflichtfeld)
|
||||
pachtbeginn: Datum YYYY-MM-DD (Pflichtfeld)
|
||||
verpachtete_flaeche: Fläche in qm (Pflichtfeld)
|
||||
pachtzins_pauschal: Jährlicher Pachtzins in EUR (Pflichtfeld)
|
||||
zahlungsweise: jaehrlich/halbjaehrlich/vierteljaehrlich/monatlich
|
||||
pachtende: Datum YYYY-MM-DD (optional)
|
||||
"""
|
||||
from stiftung.models import Land, LandVerpachtung, Paechter
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
try:
|
||||
land = Land.objects.get(id=land_id)
|
||||
except Land.DoesNotExist:
|
||||
return format_result({"fehler": f"Länderei {land_id} nicht gefunden"})
|
||||
|
||||
try:
|
||||
paechter = Paechter.objects.get(id=paechter_id)
|
||||
except Paechter.DoesNotExist:
|
||||
return format_result({"fehler": f"Pächter {paechter_id} nicht gefunden"})
|
||||
|
||||
if LandVerpachtung.objects.filter(vertragsnummer=vertragsnummer).exists():
|
||||
return format_result({"fehler": f"Vertragsnummer '{vertragsnummer}' existiert bereits"})
|
||||
|
||||
kwargs = {
|
||||
"land": land,
|
||||
"paechter": paechter,
|
||||
"vertragsnummer": vertragsnummer,
|
||||
"pachtbeginn": pachtbeginn,
|
||||
"verpachtete_flaeche": verpachtete_flaeche,
|
||||
"pachtzins_pauschal": pachtzins_pauschal,
|
||||
"zahlungsweise": zahlungsweise,
|
||||
"status": "aktiv",
|
||||
}
|
||||
if pachtende:
|
||||
kwargs["pachtende"] = pachtende
|
||||
|
||||
verpachtung = LandVerpachtung.objects.create(**kwargs)
|
||||
name = f"{land} – {paechter}"
|
||||
log_mcp_create(role, "verpachtung", str(verpachtung.id), name)
|
||||
return format_result({"erfolg": True, "id": str(verpachtung.id)})
|
||||
|
||||
|
||||
def paechter_anlegen(
|
||||
vorname: str,
|
||||
nachname: str,
|
||||
email: str = "",
|
||||
telefon: str = "",
|
||||
ort: str = "",
|
||||
plz: str = "",
|
||||
strasse: str = "",
|
||||
personentyp: str = "natuerlich",
|
||||
) -> str:
|
||||
"""
|
||||
Legt einen neuen Pächter an.
|
||||
|
||||
Args:
|
||||
vorname: Vorname (Pflichtfeld)
|
||||
nachname: Nachname (Pflichtfeld)
|
||||
email: E-Mail (optional)
|
||||
telefon: Telefon (optional)
|
||||
ort: Ort (optional)
|
||||
plz: Postleitzahl (optional)
|
||||
strasse: Straße (optional)
|
||||
personentyp: natuerlich/gesellschaft
|
||||
"""
|
||||
from stiftung.models import Paechter
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
kwargs = {
|
||||
"vorname": vorname.strip(),
|
||||
"nachname": nachname.strip(),
|
||||
"personentyp": personentyp,
|
||||
}
|
||||
for field, value in [("email", email), ("telefon", telefon), ("ort", ort), ("plz", plz), ("strasse", strasse)]:
|
||||
if value:
|
||||
kwargs[field] = value
|
||||
|
||||
paechter = Paechter.objects.create(**kwargs)
|
||||
name = f"{vorname} {nachname}"
|
||||
log_mcp_create(role, "paechter", str(paechter.id), name)
|
||||
return format_result({"erfolg": True, "id": str(paechter.id), "name": name})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Verwaltungskosten
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def verwaltungskosten_erfassen(
|
||||
bezeichnung: str,
|
||||
kategorie: str,
|
||||
betrag: float,
|
||||
datum: str,
|
||||
lieferant_firma: str = "",
|
||||
rechnungsnummer: str = "",
|
||||
status: str = "geplant",
|
||||
) -> str:
|
||||
"""
|
||||
Erfasst eine neue Verwaltungskosten-Position.
|
||||
|
||||
Args:
|
||||
bezeichnung: Bezeichnung (Pflichtfeld)
|
||||
kategorie: rechnung_intern/bueroausstattung/fahrtkosten/porto/telefon_internet/
|
||||
software/beratung/versicherung/steuerberatung/bankgebuehren/sonstiges
|
||||
betrag: Betrag in EUR (Pflichtfeld)
|
||||
datum: Datum YYYY-MM-DD (Pflichtfeld)
|
||||
lieferant_firma: Lieferant oder Firma (optional)
|
||||
rechnungsnummer: Rechnungsnummer (optional)
|
||||
status: geplant/bestellt/erhalten/in_bearbeitung/bezahlt/storniert
|
||||
"""
|
||||
from stiftung.models import Verwaltungskosten
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
vk = Verwaltungskosten.objects.create(
|
||||
bezeichnung=bezeichnung,
|
||||
kategorie=kategorie,
|
||||
betrag=betrag,
|
||||
datum=datum,
|
||||
lieferant_firma=lieferant_firma,
|
||||
rechnungsnummer=rechnungsnummer,
|
||||
status=status,
|
||||
)
|
||||
|
||||
log_mcp_create(role, "verwaltungskosten", str(vk.id), bezeichnung)
|
||||
return format_result({"erfolg": True, "id": str(vk.id), "bezeichnung": bezeichnung})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Termine
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def termin_anlegen(
|
||||
titel: str,
|
||||
datum: str,
|
||||
kategorie: str = "termin",
|
||||
prioritaet: str = "normal",
|
||||
beschreibung: str = "",
|
||||
uhrzeit: str = "",
|
||||
ganztags: bool = True,
|
||||
destinataer_id: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Legt einen neuen Kalendertermin an.
|
||||
|
||||
Args:
|
||||
titel: Titel des Termins (Pflichtfeld)
|
||||
datum: Datum YYYY-MM-DD (Pflichtfeld)
|
||||
kategorie: termin/zahlung/deadline/geburtstag/vertrag/pruefung/sonstiges
|
||||
prioritaet: niedrig/normal/hoch/kritisch
|
||||
beschreibung: Beschreibung (optional)
|
||||
uhrzeit: Uhrzeit HH:MM (optional)
|
||||
ganztags: Ganztägig (Standard: True)
|
||||
destinataer_id: UUID eines zugehörigen Destinatärs (optional)
|
||||
"""
|
||||
from stiftung.models import Destinataer, StiftungsKalenderEintrag
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
kwargs = {
|
||||
"titel": titel,
|
||||
"datum": datum,
|
||||
"kategorie": kategorie,
|
||||
"prioritaet": prioritaet,
|
||||
"beschreibung": beschreibung,
|
||||
"ganztags": ganztags,
|
||||
}
|
||||
if uhrzeit:
|
||||
kwargs["uhrzeit"] = uhrzeit
|
||||
kwargs["ganztags"] = False
|
||||
if destinataer_id:
|
||||
try:
|
||||
kwargs["destinataer"] = Destinataer.objects.get(id=destinataer_id)
|
||||
except Destinataer.DoesNotExist:
|
||||
return format_result({"fehler": f"Destinatär {destinataer_id} nicht gefunden"})
|
||||
|
||||
termin = StiftungsKalenderEintrag.objects.create(**kwargs)
|
||||
log_mcp_create(role, "system", str(termin.id), titel)
|
||||
return format_result({"erfolg": True, "id": str(termin.id), "titel": titel, "datum": datum})
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Dokument verknüpfen
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def dokument_verknuepfen(
|
||||
dokument_id: str,
|
||||
land_id: str = "",
|
||||
paechter_id: str = "",
|
||||
destinataer_id: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Verknüpft ein vorhandenes Dokument mit einer Länderei, einem Pächter oder Destinatär.
|
||||
|
||||
Args:
|
||||
dokument_id: UUID des Dokuments (Pflichtfeld)
|
||||
land_id: UUID der Länderei (optional)
|
||||
paechter_id: UUID des Pächters (optional)
|
||||
destinataer_id: UUID des Destinatärs (optional)
|
||||
"""
|
||||
from stiftung.models import DokumentDatei
|
||||
|
||||
role = _require_write_role()
|
||||
|
||||
try:
|
||||
dokument = DokumentDatei.objects.get(id=dokument_id)
|
||||
except DokumentDatei.DoesNotExist:
|
||||
return format_result({"fehler": f"Dokument {dokument_id} nicht gefunden"})
|
||||
|
||||
changes = {}
|
||||
update_fields = []
|
||||
|
||||
if land_id:
|
||||
from stiftung.models import Land
|
||||
try:
|
||||
land = Land.objects.get(id=land_id)
|
||||
dokument.land = land
|
||||
update_fields.append("land")
|
||||
changes["land"] = {"neu": str(land)}
|
||||
except Land.DoesNotExist:
|
||||
return format_result({"fehler": f"Länderei {land_id} nicht gefunden"})
|
||||
|
||||
if paechter_id:
|
||||
from stiftung.models import Paechter
|
||||
try:
|
||||
paechter = Paechter.objects.get(id=paechter_id)
|
||||
dokument.paechter = paechter
|
||||
update_fields.append("paechter")
|
||||
changes["paechter"] = {"neu": str(paechter)}
|
||||
except Paechter.DoesNotExist:
|
||||
return format_result({"fehler": f"Pächter {paechter_id} nicht gefunden"})
|
||||
|
||||
if destinataer_id:
|
||||
from stiftung.models import Destinataer
|
||||
try:
|
||||
dest = Destinataer.objects.get(id=destinataer_id)
|
||||
dokument.destinataer = dest
|
||||
update_fields.append("destinataer")
|
||||
changes["destinataer"] = {"neu": f"{dest.vorname} {dest.nachname}"}
|
||||
except Destinataer.DoesNotExist:
|
||||
return format_result({"fehler": f"Destinatär {destinataer_id} nicht gefunden"})
|
||||
|
||||
if not update_fields:
|
||||
return format_result({"fehler": "Keine Verknüpfung angegeben (land_id, paechter_id oder destinataer_id)"})
|
||||
|
||||
dokument.save(update_fields=update_fields)
|
||||
log_mcp_update(role, "dokumentlink", str(dokument.id), dokument.titel, changes)
|
||||
return format_result({"erfolg": True, "id": str(dokument.id), "verknuepft_mit": list(changes.keys())})
|
||||
Reference in New Issue
Block a user