From 61270e6d00579224af8c7c24f6d22aebe0664f9b Mon Sep 17 00:00:00 2001 From: Scarriffle Date: Sun, 26 Jul 2026 20:49:08 +0200 Subject: [PATCH] =?UTF-8?q?Backend:=20Kaufpreis=20+=20Belege=20an=20Einzel?= =?UTF-8?q?st=C3=BCcken,=20PDF-Garantie-Vorschlag?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Item: price_cents + currency (Kaufpreis rappen-/centgenau). Migration ergaenzt. - Neue Tabelle item_documents (mehrere Belege je Stueck, PDF oder Bild); Blob wird verzoegert geladen, damit Item-Listen leicht bleiben. - Endpunkte: Upload (POST), Ansehen/Download (GET), Loeschen (DELETE) je Stueck. Dateiname beim Download gehaertet (keine Header-Injektion). - services/warranty.py: schaetzt aus PDF-Text lokal ein Garantieende (Zeitraum + Kaufdatum, oder Datum nahe Garantie-Stichwort); pypdf ergaenzt. Der Upload liefert den Vorschlag zurueck. 10 neue Tests, Suite 136 gruen. Co-Authored-By: Claude Opus 4.8 --- backend/app/main.py | 3 + backend/app/models.py | 27 +++++++ backend/app/routers/items.py | 133 ++++++++++++++++++++++++++++++- backend/app/schemas.py | 21 +++++ backend/app/services/warranty.py | 126 +++++++++++++++++++++++++++++ backend/requirements.txt | 1 + backend/tests/test_items.py | 55 +++++++++++++ backend/tests/test_warranty.py | 39 +++++++++ 8 files changed, 402 insertions(+), 3 deletions(-) create mode 100644 backend/app/services/warranty.py create mode 100644 backend/tests/test_items.py create mode 100644 backend/tests/test_warranty.py diff --git a/backend/app/main.py b/backend/app/main.py index f0d0276..6e6fa00 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -85,6 +85,9 @@ def _ensure_schema() -> None: # Einzelstück-Verwaltung (Items mit UID/QR) je Produkt. "ALTER TABLE products ADD COLUMN IF NOT EXISTS individual BOOLEAN " "NOT NULL DEFAULT FALSE", + # Kaufpreis (in Rappen/Cent) und Währung am Einzelstück. + "ALTER TABLE items ADD COLUMN IF NOT EXISTS price_cents INTEGER", + "ALTER TABLE items ADD COLUMN IF NOT EXISTS currency VARCHAR(3)", ] with engine.begin() as conn: for stmt in stmts: diff --git a/backend/app/models.py b/backend/app/models.py index a470c5e..98b4a17 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -530,8 +530,35 @@ class Item(Base): acquired_on: Mapped[date | None] = mapped_column(Date, nullable=True) # gekauft am warranty_until: Mapped[date | None] = mapped_column(Date, nullable=True) # Garantie bis note: Mapped[str | None] = mapped_column(String(255), nullable=True) # Notiz/Zustand + # Kaufpreis in kleinster Einheit (Rappen/Cent), damit keine Float-Rundung + # entsteht; Währung getrennt (z.B. CHF, EUR). + price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True) + currency: Mapped[str | None] = mapped_column(String(3), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) product: Mapped[Product] = relationship() location: Mapped[Location | None] = relationship() shop: Mapped[Shop | None] = relationship() + documents: Mapped[list["ItemDocument"]] = relationship( + cascade="all, delete-orphan", order_by="ItemDocument.id" + ) + + +class ItemDocument(Base): + """Beleg zu einem Einzelstück: Rechnung oder Garantieschein (PDF oder Bild). + + Eigene Tabelle wie :class:`ProductImage`, damit die Binärdaten nicht bei + jeder Item-Abfrage mitgezogen werden. Mehrere Belege je Stück sind erlaubt. + """ + + __tablename__ = "item_documents" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + item_id: Mapped[int] = mapped_column( + ForeignKey("items.id", ondelete="CASCADE"), nullable=False, index=True + ) + filename: Mapped[str] = mapped_column(String(255), nullable=False) + content_type: Mapped[str] = mapped_column(String(64), nullable=False) + # Nur bei Bedarf laden (Download): sonst zoege jede Item-Liste die Belege mit. + data: Mapped[bytes] = mapped_column(LargeBinary, nullable=False, deferred=True) + uploaded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) diff --git a/backend/app/routers/items.py b/backend/app/routers/items.py index c6634a0..84b0586 100644 --- a/backend/app/routers/items.py +++ b/backend/app/routers/items.py @@ -5,17 +5,40 @@ Ein Item ist ein physisches Exemplar mit eigener UID/QR und eigenen Angaben Gegenstands-Produkte mit ``Product.individual``. """ -from fastapi import APIRouter, Depends, HTTPException, status +import re + +from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status +from fastapi.responses import Response from sqlalchemy.orm import Session from ..database import get_db from ..deps import get_current_user, require_admin -from ..models import Item, Location, Movement, MovementType, Product, Shop, User -from ..schemas import ItemCreate, ItemOut, ItemRemove, ItemUpdate +from ..models import ( + Item, + ItemDocument, + Location, + Movement, + MovementType, + Product, + Shop, + User, +) +from ..schemas import ( + ItemCreate, + ItemDocumentUploadOut, + ItemOut, + ItemRemove, + ItemUpdate, +) from ..services.items import generate_uid, item_to_out +from ..services.warranty import extract_pdf_text, guess_warranty_until router = APIRouter(tags=["items"]) +# Belege sind meist kleine PDFs; die Grenze schuetzt vor Ausreissern und passt +# zur nginx-Grenze (client_max_body_size 12m). +DOC_MAX_BYTES = 10 * 1024 * 1024 + def _product_or_404(db: Session, product_id: int) -> Product: product = db.get(Product, product_id) @@ -72,6 +95,8 @@ def create_items( acquired_on=payload.acquired_on, warranty_until=payload.warranty_until, note=payload.note, + price_cents=payload.price_cents, + currency=payload.currency, ) db.add(item) db.flush() @@ -156,3 +181,105 @@ def delete_item( """Einzelstück ohne Grund löschen (Korrektur).""" db.delete(_item_or_404(db, item_id)) db.commit() + + +# -------------------------------------------------------------------------- +# Belege (Rechnung/Garantieschein) je Einzelstück +# -------------------------------------------------------------------------- +def _document_or_404(db: Session, item_id: int, doc_id: int) -> ItemDocument: + doc = ( + db.query(ItemDocument) + .filter(ItemDocument.id == doc_id, ItemDocument.item_id == item_id) + .first() + ) + if doc is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Beleg nicht gefunden") + return doc + + +def _safe_filename(name: str | None) -> str: + """Steuerzeichen/Anführungszeichen raus – sonst Header-Injektion im Download.""" + cleaned = re.sub(r'[\r\n"]', "", (name or "").strip()) + return cleaned[:255] or "beleg" + + +@router.post( + "/items/{item_id}/documents", + response_model=ItemDocumentUploadOut, + status_code=status.HTTP_201_CREATED, +) +async def upload_item_document( + item_id: int, + file: UploadFile = File(...), + db: Session = Depends(get_db), + _: User = Depends(require_admin), +) -> ItemDocumentUploadOut: + """Beleg (PDF oder Bild) hochladen. Bei PDF wird ein Garantieende vorgeschlagen.""" + item = _item_or_404(db, item_id) + data = await file.read() + if not data: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "Leere Datei") + if len(data) > DOC_MAX_BYTES: + raise HTTPException( + status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + f"Datei zu groß (max. {DOC_MAX_BYTES // (1024 * 1024)} MB).", + ) + content_type = file.content_type or "application/octet-stream" + if content_type != "application/pdf" and not content_type.startswith("image/"): + raise HTTPException( + status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "Nur PDF oder Bild erlaubt." + ) + + doc = ItemDocument( + item_id=item.id, + filename=_safe_filename(file.filename), + content_type=content_type, + data=data, + ) + db.add(doc) + db.commit() + db.refresh(doc) + + # Garantieende nur aus PDFs schätzen (Bilder haben keine Textebene). + suggestion = None + if content_type == "application/pdf": + suggestion = guess_warranty_until( + extract_pdf_text(data), acquired_on=item.acquired_on + ) + + return ItemDocumentUploadOut( + id=doc.id, + filename=doc.filename, + content_type=doc.content_type, + uploaded_at=doc.uploaded_at, + suggested_warranty_until=suggestion, + ) + + +@router.get("/items/{item_id}/documents/{doc_id}") +def get_item_document( + item_id: int, + doc_id: int, + db: Session = Depends(get_db), + _: User = Depends(get_current_user), +) -> Response: + """Beleg anzeigen/herunterladen.""" + doc = _document_or_404(db, item_id, doc_id) + return Response( + content=doc.data, + media_type=doc.content_type, + headers={"Content-Disposition": f'inline; filename="{doc.filename}"'}, + ) + + +@router.delete( + "/items/{item_id}/documents/{doc_id}", status_code=status.HTTP_204_NO_CONTENT +) +def delete_item_document( + item_id: int, + doc_id: int, + db: Session = Depends(get_db), + _: User = Depends(require_admin), +) -> None: + db.delete(_document_or_404(db, item_id, doc_id)) + db.commit() diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 9e5de7e..e455838 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -512,6 +512,8 @@ class ItemCreate(BaseModel): acquired_on: date | None = None warranty_until: date | None = None note: str | None = Field(default=None, max_length=255) + price_cents: int | None = Field(default=None, ge=0) + currency: str | None = Field(default=None, max_length=3) class ItemUpdate(BaseModel): @@ -520,6 +522,8 @@ class ItemUpdate(BaseModel): acquired_on: date | None = None warranty_until: date | None = None note: str | None = Field(default=None, max_length=255) + price_cents: int | None = Field(default=None, ge=0) + currency: str | None = Field(default=None, max_length=3) class ItemRemove(BaseModel): @@ -527,6 +531,20 @@ class ItemRemove(BaseModel): note: str | None = Field(default=None, max_length=255) +class ItemDocumentOut(BaseModel): + """Beleg-Metadaten (ohne Binärdaten).""" + model_config = ConfigDict(from_attributes=True) + id: int + filename: str + content_type: str + uploaded_at: datetime + + +class ItemDocumentUploadOut(ItemDocumentOut): + """Antwort nach dem Upload – mit dem aus dem PDF vorgeschlagenen Garantieende.""" + suggested_warranty_until: date | None = None + + class ItemOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int @@ -539,10 +557,13 @@ class ItemOut(BaseModel): acquired_on: date | None = None warranty_until: date | None = None note: str | None = None + price_cents: int | None = None + currency: str | None = None created_at: datetime # Für QR-Auflösung/Anzeige mitgeliefert: product_name: str | None = None product_brand: str | None = None + documents: list[ItemDocumentOut] = [] # ---- Views ---- diff --git a/backend/app/services/warranty.py b/backend/app/services/warranty.py new file mode 100644 index 0000000..fcd9867 --- /dev/null +++ b/backend/app/services/warranty.py @@ -0,0 +1,126 @@ +"""Aus einem Beleg-PDF ein mögliches Garantieende schätzen – lokal, ohne KI. + +Zwei Wege, in dieser Reihenfolge: +1. Garantiezeitraum ("24 Monate", "2 Jahre") + Kaufdatum (aus dem Beleg oder vom + Einzelstück) → Enddatum ausrechnen. +2. Ein Datum in der Nähe eines Garantie-Stichworts. + +Bewusst konservativ: lieber nichts vorschlagen als etwas Falsches – das Ergebnis +ist nur ein Vorschlag, den der Nutzer bestätigt. Reine Scan-PDFs ohne Textebene +liefern keinen Text und damit keinen Vorschlag (dafür bräuchte es OCR). +""" + +from __future__ import annotations + +import io +import re +from datetime import date + +from dateutil.relativedelta import relativedelta + +_KEYWORD_RE = re.compile(r"garantie|gew[äa]hrleistung|warranty", re.IGNORECASE) +_PERIOD_RE = re.compile(r"(\d{1,3})\s*(jahr|monat|year|month)", re.IGNORECASE) +_PURCHASE_KW = re.compile( + r"rechnungsdatum|rechnung|kaufdatum|bestelldatum|bestellt|belegdatum|" + r"datum|invoice|order date|purchase|receipt", + re.IGNORECASE, +) +# 2024-03-31 oder 31.03.2024 / 31/03/24 +_DATE_RE = re.compile( + r"(\d{4})-(\d{1,2})-(\d{1,2})|(\d{1,2})[.\/](\d{1,2})[.\/](\d{2,4})" +) + +_WINDOW = 60 # Zeichen um ein Stichwort, in denen Zahl/Datum als zugehörig gelten + + +def extract_pdf_text(data: bytes) -> str: + """Text aus einem PDF ziehen. Schlägt es fehl (kaputt, reiner Scan), "".""" + try: + from pypdf import PdfReader + + reader = PdfReader(io.BytesIO(data)) + return "\n".join((page.extract_text() or "") for page in reader.pages) + except Exception: + return "" + + +def _match_to_date(m: re.Match) -> date | None: + g = m.groups() + try: + if g[0] is not None: # ISO yyyy-mm-dd + y, mo, d = int(g[0]), int(g[1]), int(g[2]) + else: # dd.mm.yyyy / dd/mm/yy + d, mo, y = int(g[3]), int(g[4]), int(g[5]) + if y < 100: + y += 2000 + return date(y, mo, d) + except (TypeError, ValueError): + return None + + +def _all_dates(text: str) -> list[tuple[int, date]]: + out: list[tuple[int, date]] = [] + for m in _DATE_RE.finditer(text): + d = _match_to_date(m) + if d is not None: + out.append((m.start(), d)) + return out + + +def _find_period(text: str) -> tuple[int, str] | None: + """(Anzahl, Einheit) in der Nähe eines Garantie-Stichworts – beide Reihenfolgen.""" + for kw in _KEYWORD_RE.finditer(text): + fenster = text[max(0, kw.start() - _WINDOW): kw.end() + _WINDOW] + p = _PERIOD_RE.search(fenster) + if p: + return int(p.group(1)), p.group(2).lower() + return None + + +def _find_purchase_date(text: str) -> date | None: + dates = _all_dates(text) + if not dates: + return None + best: date | None = None + best_dist = 10**9 + for kw in _PURCHASE_KW.finditer(text): + for pos, d in dates: + dist = abs(pos - kw.start()) + if dist <= _WINDOW and dist < best_dist: + best, best_dist = d, dist + # Kein Kauf-Stichwort getroffen: das früheste Datum liegt am ehesten vor dem + # Garantieende. + return best if best is not None else min(d for _, d in dates) + + +def _find_date_near_keyword(text: str) -> date | None: + dates = _all_dates(text) + best: date | None = None + best_dist = 10**9 + for kw in _KEYWORD_RE.finditer(text): + for pos, d in dates: + dist = abs(pos - kw.start()) + if dist <= _WINDOW and dist < best_dist: + best, best_dist = d, dist + return best + + +def _add_period(base: date, n: int, unit: str) -> date: + if unit.startswith(("jahr", "year")): + return base + relativedelta(years=n) + return base + relativedelta(months=n) + + +def guess_warranty_until(text: str, acquired_on: date | None = None) -> date | None: + """Bestes rät fürs Garantieende oder ``None``. + + ``acquired_on`` ist das am Einzelstück hinterlegte Kaufdatum – es hat Vorrang + vor einem im Beleg gefundenen Datum. + """ + if not text: + return None + period = _find_period(text) + base = acquired_on or _find_purchase_date(text) + if period and base: + return _add_period(base, period[0], period[1]) + return _find_date_near_keyword(text) diff --git a/backend/requirements.txt b/backend/requirements.txt index dfc412b..793b892 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -9,4 +9,5 @@ bcrypt==4.2.1 python-multipart==0.0.20 httpx==0.28.1 python-dateutil==2.9.0.post0 +pypdf==5.1.0 pytest==8.3.4 diff --git a/backend/tests/test_items.py b/backend/tests/test_items.py new file mode 100644 index 0000000..9ff0e36 --- /dev/null +++ b/backend/tests/test_items.py @@ -0,0 +1,55 @@ +"""Einzelstücke: Kaufpreis und Belege (Metadaten in der Ausgabe, Dateinamen-Härtung).""" + +import pytest + +from app.models import Category, CategoryTracking, Item, ItemDocument, Product, Role, User +from app.routers.items import _safe_filename, create_items, get_item +from app.schemas import ItemCreate + + +@pytest.fixture() +def admin(db): + person = User(username="admin", password_hash="x", role=Role.admin) + db.add(person) + db.commit() + db.refresh(person) + return person + + +def _product(db): + cat = Category(name="Elektronik", tracking=CategoryTracking.object.value) + db.add(cat) + db.flush() + product = Product(name="Kamera", category_id=cat.id, individual=True) + db.add(product) + db.commit() + db.refresh(product) + return product + + +def test_item_kaufpreis_round_trip(db, admin): + product = _product(db) + out = create_items( + product.id, ItemCreate(count=1, price_cents=49900, currency="CHF"), db=db, _=admin + ) + assert out[0].price_cents == 49900 + assert out[0].currency == "CHF" + + +def test_item_out_listet_belege(db, admin): + product = _product(db) + item_id = create_items(product.id, ItemCreate(count=1), db=db, _=admin)[0].id + db.add(ItemDocument(item_id=item_id, filename="rechnung.pdf", + content_type="application/pdf", data=b"%PDF-1.4 x")) + db.commit() + + fresh = get_item(item_id, db=db, _=admin) + assert len(fresh.documents) == 1 + assert fresh.documents[0].filename == "rechnung.pdf" + assert fresh.documents[0].content_type == "application/pdf" + + +def test_safe_filename_entfernt_header_zeichen(): + assert _safe_filename('a"b\r\nc.pdf') == "abc.pdf" + assert _safe_filename(" ") == "beleg" + assert _safe_filename(None) == "beleg" diff --git a/backend/tests/test_warranty.py b/backend/tests/test_warranty.py new file mode 100644 index 0000000..2b86943 --- /dev/null +++ b/backend/tests/test_warranty.py @@ -0,0 +1,39 @@ +"""Lokale Garantieende-Schätzung aus Belegtext (ohne echtes PDF).""" + +from datetime import date + +from app.services.warranty import guess_warranty_until + + +def test_zeitraum_plus_kaufdatum_vom_stueck(): + # Beleg nennt nur den Zeitraum; das Kaufdatum kommt vom Einzelstück. + text = "Vielen Dank für Ihren Einkauf. 24 Monate Garantie auf dieses Gerät." + assert guess_warranty_until(text, acquired_on=date(2024, 3, 31)) == date(2026, 3, 31) + + +def test_jahre_werden_verstanden(): + text = "2 Jahre Gewährleistung" + assert guess_warranty_until(text, acquired_on=date(2023, 1, 15)) == date(2025, 1, 15) + + +def test_zeitraum_plus_rechnungsdatum_aus_dem_beleg(): + text = "Rechnungsdatum: 05.06.2024\nGarantie: 12 Monate" + assert guess_warranty_until(text) == date(2025, 6, 5) + + +def test_warranty_keyword_before_period_english(): + text = "Warranty period is 36 months from purchase." + assert guess_warranty_until(text, acquired_on=date(2022, 2, 1)) == date(2025, 2, 1) + + +def test_explizites_datum_neben_stichwort(): + text = "Garantie gültig bis 31.12.2027." + assert guess_warranty_until(text) == date(2027, 12, 31) + + +def test_ohne_hinweise_kein_vorschlag(): + assert guess_warranty_until("Nur ein Kassenbon ohne alles.") is None + + +def test_leerer_text_ist_none(): + assert guess_warranty_until("", acquired_on=date(2024, 1, 1)) is None