Scarriffle/Vorrania
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backend: Beleg-Analyse um Kaufdatum + Shop; Analyse-Endpoint

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-27 09:30:35 +02:00
parent 729dfe766b
commit bbf10b36c6
4 changed files with 141 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
backend/app/routers/items.py
View File

@@ -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,

12
backend/app/schemas.py
View File

@@ -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):

51
backend/app/services/warranty.py
View File

@@ -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
# --------------------------------------------------------------------------

24
backend/tests/test_warranty.py
View File

@@ -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)