From bbf10b36c6d251fff9c4f5101cc655449a2d7296 Mon Sep 17 00:00:00 2001 From: Scarriffle Date: Mon, 27 Jul 2026 09:30:35 +0200 Subject: [PATCH] Backend: Beleg-Analyse um Kaufdatum + Shop; Analyse-Endpoint - guess_acquired_on (Kaufdatum) und guess_shop (bekannter Shop erkannt -> id; sonst Kandidatenname zum Anlegen) ergaenzt. - Upload liefert zusaetzlich suggested_acquired_on / shop_id / shop_name. - Neuer POST /items/analyze-document: analysiert einen Beleg OHNE zu speichern (zum Vorbefuellen beim Anlegen). Gemeinsamer Helfer _doc_suggestions. - 6 neue Tests; Suite 150 gruen. Co-Authored-By: Claude Opus 4.8 --- backend/app/routers/items.py | 82 ++++++++++++++++++++++---------- backend/app/schemas.py | 12 ++++- backend/app/services/warranty.py | 51 ++++++++++++++++++++ backend/tests/test_warranty.py | 24 ++++++++++ 4 files changed, 141 insertions(+), 28 deletions(-) diff --git a/backend/app/routers/items.py b/backend/app/routers/items.py index 2d57e6b..c7d3323 100644 --- a/backend/app/routers/items.py +++ b/backend/app/routers/items.py @@ -24,6 +24,7 @@ from ..models import ( User, ) from ..schemas import ( + DocumentSuggestions, ItemCreate, ItemDocumentUploadOut, ItemOut, @@ -33,8 +34,10 @@ from ..schemas import ( from ..services.items import generate_uid, item_to_out from ..services.warranty import ( extract_pdf_text, + guess_acquired_on, guess_price_candidates, guess_price_cents, + guess_shop, guess_warranty_until, ) @@ -208,20 +211,8 @@ def _safe_filename(name: str | None) -> str: 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() +def _read_document(file: UploadFile, data: bytes) -> str: + """Validiert Größe/Typ eines hochgeladenen Belegs und gibt den Content-Type.""" if not data: raise HTTPException(status.HTTP_400_BAD_REQUEST, "Leere Datei") if len(data) > DOC_MAX_BYTES: @@ -234,6 +225,43 @@ async def upload_item_document( raise HTTPException( status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "Nur PDF oder Bild erlaubt." ) + return content_type + + +def _doc_suggestions( + db: Session, data: bytes, content_type: str, item_acquired_on=None +) -> DocumentSuggestions: + """Garantie, Preis, Kaufdatum und Shop aus einem PDF schätzen (Bilder: leer).""" + if content_type != "application/pdf": + return DocumentSuggestions() + text = extract_pdf_text(data) + shop_id, shop_name = guess_shop(text, [(s.id, s.name) for s in db.query(Shop).all()]) + return DocumentSuggestions( + suggested_warranty_until=guess_warranty_until(text, acquired_on=item_acquired_on), + suggested_price_cents=guess_price_cents(text), + suggested_price_candidates=guess_price_candidates(text), + suggested_acquired_on=guess_acquired_on(text), + suggested_shop_id=shop_id, + suggested_shop_name=shop_name, + ) + + +@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 werden Garantie, Preis, Kaufdatum + und Shop vorgeschlagen.""" + item = _item_or_404(db, item_id) + data = await file.read() + content_type = _read_document(file, data) doc = ItemDocument( item_id=item.id, @@ -245,26 +273,28 @@ async def upload_item_document( db.commit() db.refresh(doc) - # Garantieende und Preis nur aus PDFs schätzen (Bilder haben keine Textebene). - warranty = price = None - candidates: list[int] = [] - if content_type == "application/pdf": - text = extract_pdf_text(data) - warranty = guess_warranty_until(text, acquired_on=item.acquired_on) - price = guess_price_cents(text) - candidates = guess_price_candidates(text) - + vorschlag = _doc_suggestions(db, data, content_type, item.acquired_on) return ItemDocumentUploadOut( id=doc.id, filename=doc.filename, content_type=doc.content_type, uploaded_at=doc.uploaded_at, - suggested_warranty_until=warranty, - suggested_price_cents=price, - suggested_price_candidates=candidates, + **vorschlag.model_dump(), ) +@router.post("/items/analyze-document", response_model=DocumentSuggestions) +async def analyze_item_document( + file: UploadFile = File(...), + db: Session = Depends(get_db), + _: User = Depends(require_admin), +) -> DocumentSuggestions: + """Beleg nur analysieren (nichts speichern) – zum Vorbefüllen beim Anlegen.""" + data = await file.read() + content_type = _read_document(file, data) + return _doc_suggestions(db, data, content_type) + + @router.get("/items/{item_id}/documents/{doc_id}") def get_item_document( item_id: int, diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 485285a..831a76a 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -557,12 +557,20 @@ class ItemDocumentOut(BaseModel): uploaded_at: datetime -class ItemDocumentUploadOut(ItemDocumentOut): - """Antwort nach dem Upload – mit aus dem PDF geschätztem Garantieende und Preis.""" +class DocumentSuggestions(BaseModel): + """Aus einem Beleg-PDF geschätzte Werte (ohne den Beleg zu speichern).""" suggested_warranty_until: date | None = None suggested_price_cents: int | None = None # Alle plausiblen Preise (bester zuerst) – zur Auswahl, falls es mehrere gibt. suggested_price_candidates: list[int] = [] + suggested_acquired_on: date | None = None + # Erkannter Shop: id = bereits hinterlegt; sonst name = Vorschlag zum Anlegen. + suggested_shop_id: int | None = None + suggested_shop_name: str | None = None + + +class ItemDocumentUploadOut(ItemDocumentOut, DocumentSuggestions): + """Antwort nach dem Upload – Beleg-Metadaten plus die geschätzten Werte.""" class ItemOut(BaseModel): diff --git a/backend/app/services/warranty.py b/backend/app/services/warranty.py index a40c4a6..bab88be 100644 --- a/backend/app/services/warranty.py +++ b/backend/app/services/warranty.py @@ -126,6 +126,57 @@ def guess_warranty_until(text: str, acquired_on: date | None = None) -> date | N return _find_date_near_keyword(text) +def guess_acquired_on(text: str) -> date | None: + """Kaufdatum aus dem Beleg (Datum nahe Rechnungs-/Kaufdatum-Stichwort).""" + return _find_purchase_date(text) if text else None + + +# -------------------------------------------------------------------------- +# Shop / Bezugsquelle aus dem Beleg +# -------------------------------------------------------------------------- +_LEGAL_RE = re.compile( + r"\b(GmbH|AG|SA|Sàrl|S\.?à r\.?l\.?|Ltd|Inc|SE|KG|OHG|e\.?K\.?|AS|BV|S\.p\.A\.)\b" +) + + +def _vendor_candidate(text: str) -> str | None: + """Bester Rate-Name des Händlers – meist im Kopf des Belegs. Best effort.""" + lines = [z.strip() for z in text.splitlines() if z.strip()] + for z in lines[:20]: + if _LEGAL_RE.search(z): + return z[:60] + for z in lines[:8]: + buchstaben = sum(c.isalpha() for c in z) + if 3 <= len(z) <= 40 and buchstaben >= 3 and not re.search( + r"rechnung|invoice|quittung|beleg|kassenbon|datum|receipt|order|bestell", + z, re.IGNORECASE, + ): + return z[:60] + return None + + +def guess_shop(text: str, shops: list[tuple[int, str]]) -> tuple[int | None, str | None]: + """(shop_id, name). + + Kommt der Name eines bereits hinterlegten Shops im Beleg vor, wird dieser + vorgeschlagen (längster Treffer gewinnt). Sonst ein Kandidatenname zum + Anlegen – oder (None, None), wenn nichts Brauchbares gefunden wird. + """ + if not text: + return (None, None) + low = text.lower() + treffer: tuple[int, str] | None = None + best_len = 0 + for sid, name in shops: + n = name.strip().lower() + if len(n) >= 3 and n in low and len(n) > best_len: + treffer = (sid, name) + best_len = len(n) + if treffer is not None: + return treffer + return (None, _vendor_candidate(text)) + + # -------------------------------------------------------------------------- # Kaufpreis aus dem Beleg schätzen # -------------------------------------------------------------------------- diff --git a/backend/tests/test_warranty.py b/backend/tests/test_warranty.py index ea7d6ac..281fbd3 100644 --- a/backend/tests/test_warranty.py +++ b/backend/tests/test_warranty.py @@ -3,8 +3,10 @@ from datetime import date from app.services.warranty import ( + guess_acquired_on, guess_price_candidates, guess_price_cents, + guess_shop, guess_warranty_until, ) @@ -86,3 +88,25 @@ def test_preis_kandidaten_bester_zuerst_ohne_dubletten(): assert kandidaten[0] == 5490 # bester Tipp (Summe) vorne assert set(kandidaten) == {5490, 4900, 590} # ohne Dubletten assert guess_price_candidates("") == [] + + +# ---- Kaufdatum & Shop ---- + +def test_kaufdatum_aus_beleg(): + assert guess_acquired_on("Rechnungsdatum: 05.06.2024") == date(2024, 6, 5) + assert guess_acquired_on("nichts hier") is None + + +def test_shop_erkennt_bekannten_namen(): + shops = [(1, "Digitec"), (2, "Galaxus")] + assert guess_shop("Rechnung von Digitec AG, Zürich", shops) == (1, "Digitec") + + +def test_shop_schlaegt_neuen_namen_vor(): + sid, name = guess_shop("ACME Electronics GmbH\nRechnung\nDatum 01.01.2024", [(1, "Galaxus")]) + assert sid is None + assert name == "ACME Electronics GmbH" + + +def test_shop_ohne_text_leer(): + assert guess_shop("", [(1, "Digitec")]) == (None, None)