feat(mangel): Mangel-Ticket-System (Roadmap Phase 2)

Eigener Workflow für Schäden/Defekte, getrennt von Fehlbestand (Mengen-
Abweichung Soll/Ist). Status: neu → in_bearbeitung → ersatzteil_bestellt/
reparatur_geplant → erledigt. entitaet_typ/entitaet_id optional (Objektposition
oder Geräteinstanz), analog zum bestehenden historie.entitaet_typ-Muster,
objekt_id immer gesetzt für Dashboard-Abfragen.

Ein offener Mangel mit prioritaet=hoch macht das Objekt "nicht einsatzbereit"
(neuer Grund mangel_kritisch_offen in dashboard.einsatzbereitschaft) -
niedrige/normale Mängel blockieren nicht.

Melden/Lesen: alle Mitarbeiter+. Status ändern: Materialverantwortliche+.
Neue Seite /maengel für alle eingeloggten Nutzer (Nav-Link "Mängel").

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KC8HYvv6UkCVYheYiTw9DD

backend/alembic/versions/0013_mangel.py

@@ -0,0 +1,62 @@
+"""Mangel-Ticket-System (Roadmap Phase 2, BOS-Ressourcenplattform-Erweiterung).
+
+Eigener Workflow für Schäden/Defekte - getrennt von Fehlbestand (Mengen-
+Abweichung Soll/Ist), da fachlich verschieden (Nutzer-Beispiel: "Blaulicht
+vorne rechts defekt" ist kein Fehlbestand). entitaet_typ/entitaet_id als
+String-Polymorphie analog historie.entitaet_typ/entitaet_id (bestehendes
+Muster, keine FK-Integrität nötig - Mangel bleibt auch dann lesbar, wenn die
+konkrete Objektposition/Geraet-Instanz später gelöscht wird). objekt_id
+zusätzlich direkt/redundant gespeichert (wie fehlbestand.objekt_id), damit die
+Einsatzbereitschafts-Abfrage (Dashboard) nicht über die polymorphe Spalte
+joinen muss.
+
+Revision ID: 0013_mangel
+Revises: 0012_personal_qualifikationen
+Create Date: 2026-09-05
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0013_mangel"
+down_revision: Union[str, None] = "0012_personal_qualifikationen"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "CREATE TYPE mangel_status AS ENUM "
+        "('neu', 'in_bearbeitung', 'ersatzteil_bestellt', 'reparatur_geplant', 'erledigt')"
+    )
+    op.execute("CREATE TYPE mangel_prioritaet AS ENUM ('niedrig', 'normal', 'hoch')")
+
+    op.execute(
+        """
+        CREATE TABLE mangel (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            objekt_id INTEGER NOT NULL REFERENCES objekt(id),
+            entitaet_typ TEXT,
+            entitaet_id TEXT,
+            beschreibung TEXT NOT NULL,
+            prioritaet mangel_prioritaet NOT NULL DEFAULT 'normal',
+            status mangel_status NOT NULL DEFAULT 'neu',
+            gemeldet_von INTEGER NOT NULL REFERENCES benutzer(id),
+            gemeldet_am TIMESTAMPTZ NOT NULL,
+            erledigt_von INTEGER REFERENCES benutzer(id),
+            erledigt_am TIMESTAMPTZ,
+            CONSTRAINT mangel_entitaet_beide_oder_keins CHECK (
+                (entitaet_typ IS NULL AND entitaet_id IS NULL)
+                OR (entitaet_typ IS NOT NULL AND entitaet_id IS NOT NULL)
+            )
+        )
+        """
+    )
+    op.execute("CREATE INDEX idx_mangel_objekt ON mangel (objekt_id)")
+    op.execute("CREATE INDEX idx_mangel_status ON mangel (status)")
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE mangel")
+    op.execute("DROP TYPE IF EXISTS mangel_prioritaet")
+    op.execute("DROP TYPE IF EXISTS mangel_status")

backend/app/api/v1/api.py

@@ -10,6 +10,7 @@ from app.api.v1.endpoints import (
     health,
     historie,
     kontrollen,
+    mangel,
     objekte,
     personal,
     stammdaten,
@@ -32,3 +33,4 @@ api_router.include_router(historie.router, tags=["historie"])
 api_router.include_router(geraet_instanz.router, tags=["geraet_instanz"])
 api_router.include_router(dashboard.router, tags=["dashboard"])
 api_router.include_router(personal.router, tags=["personal"])
+api_router.include_router(mangel.router, tags=["mangel"])

backend/app/api/v1/endpoints/mangel.py

@@ -0,0 +1,85 @@
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import require_roles
+from app.db.session import get_db
+from app.models.auth import RolleTyp
+from app.models.mangel import Mangel, MangelStatus
+from app.models.objekt import Objekt
+from app.schemas.mangel import MangelCreate, MangelRead, MangelStatusUpdate
+from app.services.mangel import erledige_mangel
+
+router = APIRouter()
+
+_mitarbeiter_plus = require_roles(
+    RolleTyp.mitarbeiter,
+    RolleTyp.materialverantwortlicher,
+    RolleTyp.leitungsverantwortlicher,
+    RolleTyp.administration,
+)
+_materialverantwortliche = require_roles(RolleTyp.administration, RolleTyp.materialverantwortlicher)
+
+
+@router.get("/maengel", response_model=list[MangelRead])
+async def liste_maengel(
+    objekt_id: int | None = None,
+    status_filter: MangelStatus | None = None,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_mitarbeiter_plus),
+) -> list[Mangel]:
+    """Prompt-Erweiterung (Mangel-Modul): jeder Mitarbeiter+ darf lesen (Mängel
+    sind für alle im Einsatzalltag relevant), Status ändern bleibt
+    Materialverantwortlichen+ vorbehalten (Melden vs. Bearbeiten getrennt)."""
+    stmt = select(Mangel)
+    if objekt_id is not None:
+        stmt = stmt.where(Mangel.objekt_id == objekt_id)
+    if status_filter is not None:
+        stmt = stmt.where(Mangel.status == status_filter)
+    stmt = stmt.order_by(Mangel.gemeldet_am.desc())
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+@router.post("/maengel", response_model=MangelRead, status_code=status.HTTP_201_CREATED)
+async def melde_mangel(
+    payload: MangelCreate, db: AsyncSession = Depends(get_db), current_user=Depends(_mitarbeiter_plus)
+) -> Mangel:
+    objekt = await db.get(Objekt, payload.objekt_id)
+    if objekt is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
+
+    mangel = Mangel(
+        objekt_id=payload.objekt_id,
+        entitaet_typ=payload.entitaet_typ,
+        entitaet_id=payload.entitaet_id,
+        beschreibung=payload.beschreibung,
+        prioritaet=payload.prioritaet,
+        gemeldet_von=current_user.id,
+        gemeldet_am=datetime.now(timezone.utc),
+    )
+    db.add(mangel)
+    await db.flush()
+    return mangel
+
+
+@router.patch("/maengel/{mangel_id}", response_model=MangelRead)
+async def aendere_mangel_status(
+    mangel_id: uuid.UUID,
+    payload: MangelStatusUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user=Depends(_materialverantwortliche),
+) -> Mangel:
+    mangel = await db.get(Mangel, mangel_id)
+    if mangel is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Mangel nicht gefunden")
+
+    if payload.status == MangelStatus.erledigt:
+        return await erledige_mangel(db, mangel=mangel, benutzer_id=current_user.id)
+
+    mangel.status = payload.status
+    await db.flush()
+    return mangel

backend/app/models/mangel.py

@@ -0,0 +1,61 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import CheckConstraint, ForeignKey, String
+from sqlalchemy.dialects.postgresql import ENUM as PgEnum, TIMESTAMP, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class MangelStatus(str, enum.Enum):
+    neu = "neu"
+    in_bearbeitung = "in_bearbeitung"
+    ersatzteil_bestellt = "ersatzteil_bestellt"
+    reparatur_geplant = "reparatur_geplant"
+    erledigt = "erledigt"
+
+
+class MangelPrioritaet(str, enum.Enum):
+    niedrig = "niedrig"
+    normal = "normal"
+    hoch = "hoch"
+
+
+mangel_status_pg = PgEnum(MangelStatus, name="mangel_status", create_type=False)
+mangel_prioritaet_pg = PgEnum(MangelPrioritaet, name="mangel_prioritaet", create_type=False)
+
+
+class Mangel(Base):
+    """Roadmap Phase 2: Schäden/Defekte als eigener Workflow, getrennt von
+    Fehlbestand (Mengen-Abweichung). entitaet_typ/entitaet_id optional und als
+    String-Polymorphie geführt (Muster wie historie.entitaet_typ/-id) -
+    verweist z.B. auf 'objektposition'/uuid oder 'geraet_instanz'/uuid, wenn
+    der Mangel eine konkrete Teilressource betrifft statt das ganze Objekt.
+    objekt_id ist immer gesetzt (auch bei feingranularem Verweis), damit die
+    Einsatzbereitschaft (Dashboard) direkt filtern kann, ohne die polymorphe
+    Spalte auflösen zu müssen."""
+
+    __tablename__ = "mangel"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    objekt_id: Mapped[int] = mapped_column(ForeignKey("objekt.id"), nullable=False)
+    entitaet_typ: Mapped[str | None] = mapped_column(String)
+    entitaet_id: Mapped[str | None] = mapped_column(String)
+    beschreibung: Mapped[str] = mapped_column(String, nullable=False)
+    prioritaet: Mapped[MangelPrioritaet] = mapped_column(
+        mangel_prioritaet_pg, nullable=False, default=MangelPrioritaet.normal
+    )
+    status: Mapped[MangelStatus] = mapped_column(mangel_status_pg, nullable=False, default=MangelStatus.neu)
+    gemeldet_von: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
+    gemeldet_am: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
+    erledigt_von: Mapped[int | None] = mapped_column(ForeignKey("benutzer.id"))
+    erledigt_am: Mapped[datetime | None] = mapped_column(TIMESTAMP(timezone=True))
+
+    __table_args__ = (
+        CheckConstraint(
+            "(entitaet_typ IS NULL AND entitaet_id IS NULL) OR (entitaet_typ IS NOT NULL AND entitaet_id IS NOT NULL)",
+            name="mangel_entitaet_beide_oder_keins",
+        ),
+    )

backend/app/schemas/mangel.py

@@ -0,0 +1,42 @@
+import uuid
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from app.models.mangel import MangelPrioritaet, MangelStatus
+
+EntitaetTyp = Literal["objektposition", "geraet_instanz"]
+
+
+class MangelCreate(BaseModel):
+    objekt_id: int
+    entitaet_typ: EntitaetTyp | None = None
+    entitaet_id: str | None = None
+    beschreibung: str
+    prioritaet: MangelPrioritaet = MangelPrioritaet.normal
+
+    @model_validator(mode="after")
+    def _entitaet_beide_oder_keins(self) -> "MangelCreate":
+        if (self.entitaet_typ is None) != (self.entitaet_id is None):
+            raise ValueError("entitaet_typ und entitaet_id müssen beide gesetzt oder beide leer sein")
+        return self
+
+
+class MangelRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    objekt_id: int
+    entitaet_typ: str | None
+    entitaet_id: str | None
+    beschreibung: str
+    prioritaet: MangelPrioritaet
+    status: MangelStatus
+    gemeldet_von: int
+    gemeldet_am: datetime
+    erledigt_von: int | None
+    erledigt_am: datetime | None
+
+
+class MangelStatusUpdate(BaseModel):
+    status: MangelStatus

backend/app/services/dashboard.py

@@ -10,6 +10,7 @@ from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
 from app.models.objekt import Objekt
 from app.models.objektposition import Objektposition
 from app.models.stammdaten import Material
+from app.services.mangel import objekt_ids_mit_kritischem_offenem_mangel
 
 # Prompt 12.2: offen/in_bearbeitung/nachgefuellt_teilweise zählen gemeinsam als "offen"
 # (technischer Substatus ist keine Erledigung, Prompt 03/20).
@@ -160,6 +161,8 @@ async def einsatzbereitschaft(db: AsyncSession) -> dict:
         .all()
     )
 
+    objekt_ids_kritischer_mangel = await objekt_ids_mit_kritischem_offenem_mangel(db)
+
     bereit = eingeschraenkt = 0
     details: list[dict] = []
     for objekt in objekte:
@@ -174,6 +177,8 @@ async def einsatzbereitschaft(db: AsyncSession) -> dict:
             gruende.append("pruefung_ueberfaellig")
         if objekt.id in objekt_ids_gesperrt:
             gruende.append("kontrolle_laeuft")
+        if objekt.id in objekt_ids_kritischer_mangel:
+            gruende.append("mangel_kritisch_offen")
 
         if not gruende and fehlbestand_ids and fehlbestand_ids <= fehlbestand_ids_genehmigt:
             eingeschraenkt += 1

backend/app/services/mangel.py

@@ -0,0 +1,33 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.mangel import Mangel, MangelPrioritaet, MangelStatus
+
+OFFENE_MANGEL_STATI = (
+    MangelStatus.neu,
+    MangelStatus.in_bearbeitung,
+    MangelStatus.ersatzteil_bestellt,
+    MangelStatus.reparatur_geplant,
+)
+
+
+async def objekt_ids_mit_kritischem_offenem_mangel(db: AsyncSession) -> set[int]:
+    """Dashboard/Einsatzbereitschaft (Nutzer-Vorgabe): nur ein offener Mangel mit
+    prioritaet=hoch macht ein Objekt "nicht einsatzbereit" - niedrige/normale
+    Mängel sind Hinweise, aber kein Einsatzbereitschafts-Blocker."""
+    result = await db.execute(
+        select(func.distinct(Mangel.objekt_id)).where(
+            Mangel.status.in_(OFFENE_MANGEL_STATI), Mangel.prioritaet == MangelPrioritaet.hoch
+        )
+    )
+    return set(result.scalars().all())
+
+
+async def erledige_mangel(db: AsyncSession, *, mangel: Mangel, benutzer_id: int) -> Mangel:
+    mangel.status = MangelStatus.erledigt
+    mangel.erledigt_von = benutzer_id
+    mangel.erledigt_am = datetime.now(timezone.utc)
+    await db.flush()
+    return mangel

backend/tests/test_mangel.py

@@ -0,0 +1,115 @@
+import pytest
+
+from tests.conftest import auth_header, login
+
+
+@pytest.mark.asyncio
+async def test_mangel_melden_und_lesen(client, objekt_mit_position, mitarbeiter_user):
+    objekt, _material = objekt_mit_position
+    token = await login(client, "mitarbeiter1")
+
+    gemeldet = await client.post(
+        "/api/v1/maengel",
+        json={"objekt_id": objekt.id, "beschreibung": "Blaulicht vorne rechts defekt", "prioritaet": "hoch"},
+        headers=auth_header(token),
+    )
+    assert gemeldet.status_code == 201
+    assert gemeldet.json()["status"] == "neu"
+
+    liste = await client.get(f"/api/v1/maengel?objekt_id={objekt.id}", headers=auth_header(token))
+    assert len(liste.json()) == 1
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_status_nicht_aendern(client, objekt_mit_position, mitarbeiter_user):
+    objekt, _material = objekt_mit_position
+    token = await login(client, "mitarbeiter1")
+    gemeldet = await client.post(
+        "/api/v1/maengel",
+        json={"objekt_id": objekt.id, "beschreibung": "Reifen platt"},
+        headers=auth_header(token),
+    )
+    mangel_id = gemeldet.json()["id"]
+
+    response = await client.patch(
+        f"/api/v1/maengel/{mangel_id}", json={"status": "in_bearbeitung"}, headers=auth_header(token)
+    )
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_materialverantwortlicher_kann_status_aendern_und_erledigen(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, _material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    gemeldet = await client.post(
+        "/api/v1/maengel",
+        json={"objekt_id": objekt.id, "beschreibung": "Sirene ausgefallen"},
+        headers=auth_header(mitarbeiter_token),
+    )
+    mangel_id = gemeldet.json()["id"]
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    bearbeitung = await client.patch(
+        f"/api/v1/maengel/{mangel_id}", json={"status": "ersatzteil_bestellt"}, headers=auth_header(verantwortlicher_token)
+    )
+    assert bearbeitung.json()["status"] == "ersatzteil_bestellt"
+
+    erledigt = await client.patch(
+        f"/api/v1/maengel/{mangel_id}", json={"status": "erledigt"}, headers=auth_header(verantwortlicher_token)
+    )
+    assert erledigt.json()["status"] == "erledigt"
+    assert erledigt.json()["erledigt_am"] is not None
+
+
+@pytest.mark.asyncio
+async def test_entitaet_typ_und_id_muessen_zusammen_gesetzt_sein(client, objekt_mit_position, mitarbeiter_user):
+    objekt, _material = objekt_mit_position
+    token = await login(client, "mitarbeiter1")
+    response = await client.post(
+        "/api/v1/maengel",
+        json={"objekt_id": objekt.id, "beschreibung": "X", "entitaet_typ": "geraet_instanz"},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_kritischer_offener_mangel_macht_objekt_nicht_einsatzbereit(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, _material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    await client.post(
+        "/api/v1/maengel",
+        json={"objekt_id": objekt.id, "beschreibung": "Bremsen defekt", "prioritaet": "hoch"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.get(
+        "/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(verantwortlicher_token)
+    )
+    body = response.json()
+    assert body["nicht_einsatzbereit"] == 1
+    assert body["details"][0]["gruende"] == ["mangel_kritisch_offen"]
+
+
+@pytest.mark.asyncio
+async def test_niedrige_prioritaet_blockiert_einsatzbereitschaft_nicht(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, _material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    await client.post(
+        "/api/v1/maengel",
+        json={"objekt_id": objekt.id, "beschreibung": "Kratzer im Lack", "prioritaet": "niedrig"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.get(
+        "/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(verantwortlicher_token)
+    )
+    assert response.json()["einsatzbereit"] == 1