feat(dokumente,identity): DOC-003 Versionierung + QR-Code-Etiketten
CI / backend-tests (push) Failing after 2m15s
CI / frontend-build (push) Successful in 28s

Dokument-Versionierung: vorgaenger_id verkettet Versionen (Migration 0027),
POST /dokumente/{id}/ersetzen legt neue Version an statt zu überschreiben,
alte Fassung bleibt als Historie erhalten. GET /dokumente/{id}/versionen
liefert die Kette (neueste zuerst), respektiert DOC-005-Zugriffsrechte.
Liste zeigt nur die jeweils aktuelle Version. Löschen der aktuellen Version
gibt automatisch den Vorgänger als neue "aktuelle" frei. Frontend:
"Ersetzen"-Button + aufklappbare Versionshistorie im DokumentePanel.

QR-Code-Etiketten: bisher nur Code128-Barcode möglich. generiere_qr_label_pdf
(reportlab-eigenes QR-Widget, keine neue Abhängigkeit) als Alternative,
?format=qr|code128 an /objekte/{id}/label.pdf und /objektpositionen/{id}/
label.pdf. AktePage bekommt "Etikett (Barcode)"/"Etikett (QR)"-Buttons
(vorher gab es dafür noch gar keinen UI-Zugang im Web-Frontend).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
2026-09-08 11:03:46 +02:00
co-authored by Claude Sonnet 5
parent d5d8c72d9a
commit 977d81ddcd
14 changed files with 543 additions and 54 deletions
+69 -4
View File
@@ -14,9 +14,11 @@ from app.services.dokument import (
DateizugrossError,
DokumentDuplikatError,
dateipfad,
ersetze_dokument,
liste_fuer_entitaet,
loesche_dokument,
speichere_dokument,
versionshistorie,
)
router = APIRouter()
@@ -40,6 +42,13 @@ def _darf_eingeschraenkte_dokumente_sehen(benutzer: Benutzer) -> bool:
return bool({RolleTyp(r) for r in benutzer.rollen_namen} & _VERANTWORTLICHE_ROLLEN)
def _pruefe_eingeschraenkten_zugriff(dokument: Dokument, current_user: Benutzer) -> None:
if dokument.dokumenttyp in EINGESCHRAENKTE_DOKUMENTTYPEN and not _darf_eingeschraenkte_dokumente_sehen(
current_user
):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Für diesen Dokumenttyp fehlt die Berechtigung")
@router.get("/dokumente", response_model=list[DokumentRead])
async def liste_dokumente(
entitaet_typ: EntitaetTyp,
@@ -108,16 +117,72 @@ async def lade_dokument_herunter(
dokument = await db.get(Dokument, dokument_id)
if dokument is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Dokument nicht gefunden")
if dokument.dokumenttyp in EINGESCHRAENKTE_DOKUMENTTYPEN and not _darf_eingeschraenkte_dokumente_sehen(
current_user
):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Für diesen Dokumenttyp fehlt die Berechtigung")
_pruefe_eingeschraenkten_zugriff(dokument, current_user)
pfad = dateipfad(dokument)
if not pfad.exists():
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Datei nicht mehr vorhanden")
return FileResponse(pfad, media_type=dokument.mime_type, filename=dokument.dateiname)
@router.post("/dokumente/{dokument_id}/ersetzen", response_model=DokumentRead, status_code=status.HTTP_201_CREATED)
async def dokument_ersetzen(
dokument_id: uuid.UUID,
beschreibung: str | None = Form(None),
datei: UploadFile = File(...),
db: AsyncSession = Depends(get_db),
current_user=Depends(_mitarbeiter_plus),
) -> Dokument:
"""DOC-003: neue Version anlegen statt zu überschreiben - entitaet_typ/
entitaet_id/dokumenttyp werden vom Vorgänger übernommen."""
altes_dokument = await db.get(Dokument, dokument_id)
if altes_dokument is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Dokument nicht gefunden")
_pruefe_eingeschraenkten_zugriff(altes_dokument, current_user)
inhalt = await datei.read()
try:
return await ersetze_dokument(
db,
altes_dokument=altes_dokument,
dateiname=datei.filename or "unbenannt",
mime_type=datei.content_type or "application/octet-stream",
inhalt=inhalt,
beschreibung=beschreibung,
hochgeladen_von=current_user.id,
)
except DateityperlaubtError as exc:
raise HTTPException(
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail=f"Dateityp nicht erlaubt: {exc}"
) from exc
except DateizugrossError as exc:
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="Datei zu groß"
) from exc
except DokumentDuplikatError as exc:
bestehendes = exc.bestehendes_dokument
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=(
f"Diese Datei wurde bereits am {bestehendes.hochgeladen_am:%d.%m.%Y %H:%M} "
f"als „{bestehendes.dateiname}“ hochgeladen."
),
) from exc
@router.get("/dokumente/{dokument_id}/versionen", response_model=list[DokumentRead])
async def dokument_versionen(
dokument_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
current_user: Benutzer = Depends(get_current_user),
) -> list[Dokument]:
"""DOC-003: komplette Versionskette (neueste zuerst, inkl. übergebenem
Dokument selbst)."""
dokument = await db.get(Dokument, dokument_id)
if dokument is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Dokument nicht gefunden")
_pruefe_eingeschraenkten_zugriff(dokument, current_user)
return await versionshistorie(db, dokument)
@router.delete("/dokumente/{dokument_id}", status_code=status.HTTP_204_NO_CONTENT)
async def entferne_dokument(
dokument_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
+23 -7
View File
@@ -1,4 +1,5 @@
import uuid
from typing import Literal
from fastapi import APIRouter, Depends, HTTPException, Response, status
from sqlalchemy import select
@@ -14,7 +15,9 @@ from app.models.objektposition import Objektposition
from app.models.stammdaten import Material, Objekttyp
from app.models.vorlage import Beladungsvorlage
from app.services import historie as historie_service
from app.services.label import generiere_code128_label_pdf
from app.services.label import generiere_code128_label_pdf, generiere_qr_label_pdf
LabelFormat = Literal["code128", "qr"]
from app.schemas.objekt import (
FahrzeugdetailsRead,
FahrzeugdetailsUpsert,
@@ -157,16 +160,26 @@ async def hole_objekt(
return _objekt_read(objekt, letzte_map.get(objekt.id))
def _erzeuge_label_pdf(code: str, klartext_zusatz: str | None, format: LabelFormat) -> bytes:
if format == "qr":
return generiere_qr_label_pdf(code, klartext_zusatz)
return generiere_code128_label_pdf(code, klartext_zusatz)
@router.get("/objekte/{objekt_id}/label.pdf")
async def objekt_label(
objekt_id: int, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
objekt_id: int,
format: LabelFormat = "code128",
db: AsyncSession = Depends(get_db),
_=Depends(get_current_user),
) -> Response:
"""Druckfertiges Code128-Label (Karte 10): Code + Objektname als Klartext-
Fallback. Vektor-PDF, skaliert verlustfrei auf jede Labelgröße beim Druck."""
"""Druckfertiges Etikett (Karte 10, QR-Variante Nutzer-Nachtrag): Code128
(Standard) oder QR-Code + Objektname als Klartext-Fallback. Vektor-PDF,
skaliert verlustfrei auf jede Labelgröße beim Druck."""
objekt = await db.get(Objekt, objekt_id)
if objekt is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
pdf = generiere_code128_label_pdf(objekt.code, objekt.name)
pdf = _erzeuge_label_pdf(objekt.code, objekt.name, format)
return Response(
content=pdf,
media_type="application/pdf",
@@ -176,7 +189,10 @@ async def objekt_label(
@router.get("/objektpositionen/{position_id}/label.pdf")
async def objektposition_label(
position_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
position_id: uuid.UUID,
format: LabelFormat = "code128",
db: AsyncSession = Depends(get_db),
_=Depends(get_current_user),
) -> Response:
"""Geräte-Etikett (Karte 10): eigener Code je Exemplar, Klartext-Zusatz ist
der Materialname (Objektname wäre bei mehreren Geräten am selben Objekt
@@ -185,7 +201,7 @@ async def objektposition_label(
if position is None or position.code is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Kein Geräte-Code vergeben")
material = await db.get(Material, position.material_id)
pdf = generiere_code128_label_pdf(position.code, material.name if material else None)
pdf = _erzeuge_label_pdf(position.code, material.name if material else None, format)
return Response(
content=pdf,
media_type="application/pdf",
+6
View File
@@ -61,3 +61,9 @@ class Dokument(Base):
beschreibung: Mapped[str | None] = mapped_column(String)
hochgeladen_von: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
hochgeladen_am: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
# DOC-003: Versionierung - zeigt auf das Dokument, das diese Version ersetzt.
# Alte Version bleibt als eigene Zeile erhalten (kein Überschreiben/Löschen),
# "aktuell" ist jedes Dokument, auf das kein anderes per vorgaenger_id zeigt.
vorgaenger_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True), ForeignKey("dokument.id"), nullable=True
)
+1
View File
@@ -26,3 +26,4 @@ class DokumentRead(BaseModel):
beschreibung: str | None
hochgeladen_von: int
hochgeladen_am: datetime
vorgaenger_id: uuid.UUID | None
+79 -2
View File
@@ -44,7 +44,7 @@ def _upload_pfad() -> Path:
return pfad
async def speichere_dokument(
async def _pruefe_und_speichere(
db: AsyncSession,
*,
entitaet_typ: str,
@@ -55,6 +55,7 @@ async def speichere_dokument(
dokumenttyp: DokumentTyp,
beschreibung: str | None,
hochgeladen_von: int,
vorgaenger_id: uuid.UUID | None,
) -> Dokument:
if mime_type not in ERLAUBTE_MIME_TYPES:
raise DateityperlaubtError(mime_type)
@@ -92,12 +93,81 @@ async def speichere_dokument(
beschreibung=beschreibung,
hochgeladen_von=hochgeladen_von,
hochgeladen_am=datetime.now(timezone.utc),
vorgaenger_id=vorgaenger_id,
)
db.add(dokument)
await db.flush()
return dokument
async def speichere_dokument(
db: AsyncSession,
*,
entitaet_typ: str,
entitaet_id: str,
dateiname: str,
mime_type: str,
inhalt: bytes,
dokumenttyp: DokumentTyp,
beschreibung: str | None,
hochgeladen_von: int,
) -> Dokument:
return await _pruefe_und_speichere(
db,
entitaet_typ=entitaet_typ,
entitaet_id=entitaet_id,
dateiname=dateiname,
mime_type=mime_type,
inhalt=inhalt,
dokumenttyp=dokumenttyp,
beschreibung=beschreibung,
hochgeladen_von=hochgeladen_von,
vorgaenger_id=None,
)
async def ersetze_dokument(
db: AsyncSession,
*,
altes_dokument: Dokument,
dateiname: str,
mime_type: str,
inhalt: bytes,
beschreibung: str | None,
hochgeladen_von: int,
) -> Dokument:
"""DOC-003: neue Version anlegen statt zu überschreiben - altes_dokument
bleibt unverändert als Historieneintrag erhalten, entitaet_typ/entitaet_id/
dokumenttyp werden von ihm übernommen (nicht neu wählbar, es ist dieselbe
Sache in neuer Fassung)."""
return await _pruefe_und_speichere(
db,
entitaet_typ=altes_dokument.entitaet_typ,
entitaet_id=altes_dokument.entitaet_id,
dateiname=dateiname,
mime_type=mime_type,
inhalt=inhalt,
dokumenttyp=altes_dokument.dokumenttyp,
beschreibung=beschreibung,
hochgeladen_von=hochgeladen_von,
vorgaenger_id=altes_dokument.id,
)
async def versionshistorie(db: AsyncSession, dokument: Dokument) -> list[Dokument]:
"""Kette rückwärts über vorgaenger_id, neueste zuerst (inkl. übergebenem
Dokument selbst als erstem Eintrag)."""
kette = [dokument]
aktuelles = dokument
while aktuelles.vorgaenger_id is not None:
vorgaenger = await db.get(Dokument, aktuelles.vorgaenger_id)
if vorgaenger is None:
break
kette.append(vorgaenger)
aktuelles = vorgaenger
return kette
async def liste_fuer_entitaet(
db: AsyncSession,
*,
@@ -106,7 +176,14 @@ async def liste_fuer_entitaet(
dokumenttyp: DokumentTyp | None = None,
darf_eingeschraenkte_sehen: bool = True,
) -> list[Dokument]:
bedingungen = [Dokument.entitaet_typ == entitaet_typ, Dokument.entitaet_id == entitaet_id]
# DOC-003: nur die jeweils aktuelle Version je Dokument zeigen - "aktuell"
# heißt, kein anderes Dokument hat vorgaenger_id auf dieses gesetzt.
ersetzte_ids = select(Dokument.vorgaenger_id).where(Dokument.vorgaenger_id.isnot(None))
bedingungen = [
Dokument.entitaet_typ == entitaet_typ,
Dokument.entitaet_id == entitaet_id,
Dokument.id.notin_(ersetzte_ids),
]
if dokumenttyp is not None:
bedingungen.append(Dokument.dokumenttyp == dokumenttyp)
if not darf_eingeschraenkte_sehen:
+42
View File
@@ -3,6 +3,9 @@ import io
from barcode import Code128
from barcode.writer import ImageWriter
from PIL import Image
from reportlab.graphics import renderPDF
from reportlab.graphics.barcode import qr
from reportlab.graphics.shapes import Drawing
from reportlab.lib.units import mm
from reportlab.lib.utils import ImageReader
from reportlab.pdfgen import canvas
@@ -55,3 +58,42 @@ def generiere_code128_label_pdf(code: str, klartext_zusatz: str | None = None) -
c.showPage()
c.save()
return pdf_puffer.getvalue()
_QR_ZIELGROESSE = 25 * mm
_QR_MINDESTBREITE = 30 * mm
def generiere_qr_label_pdf(code: str, klartext_zusatz: str | None = None) -> bytes:
"""Alternative zum Code128-Label (Nutzer-Wunsch): QR-Code statt 1D-Barcode -
robuster bei teilbeschädigtem Etikett (Fehlerkorrektur), von praktisch jeder
Handy-Kamera nativ lesbar, kompakter am Objekt anbringbar. Gleicher
Klartext-Code als Inhalt wie beim Code128-Label, damit beide Varianten
denselben Scan-Lookup (/objekte/code/{code}) treffen."""
qr_code = qr.QrCodeWidget(code)
qr_code.barWidth = _QR_ZIELGROESSE
qr_code.barHeight = _QR_ZIELGROESSE
bounds = qr_code.getBounds()
qr_breite_pt = bounds[2] - bounds[0]
qr_hoehe_pt = bounds[3] - bounds[1]
zeichnung = Drawing(qr_breite_pt, qr_hoehe_pt, transform=[1, 0, 0, 1, -bounds[0], -bounds[1]])
zeichnung.add(qr_code)
label_breite = max(qr_breite_pt + 2 * _RAND, _QR_MINDESTBREITE)
zusatz_hoehe = 14 if klartext_zusatz else 0
label_hoehe = qr_hoehe_pt + zusatz_hoehe + 2 * _RAND
pdf_puffer = io.BytesIO()
c = canvas.Canvas(pdf_puffer, pagesize=(label_breite, label_hoehe))
zeichnung_x = (label_breite - qr_breite_pt) / 2
zeichnung_y = label_hoehe - qr_hoehe_pt - _RAND
renderPDF.draw(zeichnung, c, zeichnung_x, zeichnung_y)
if klartext_zusatz:
c.setFont("Helvetica", 9)
c.drawCentredString(label_breite / 2, _RAND - 2, klartext_zusatz)
c.showPage()
c.save()
return pdf_puffer.getvalue()