Backend: Kaufpreis + Belege an Einzelstücken, PDF-Garantie-Vorschlag
- 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 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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 ----
|
||||
|
||||
126
backend/app/services/warranty.py
Normal file
126
backend/app/services/warranty.py
Normal file
@@ -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)
|
||||
@@ -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
|
||||
|
||||
55
backend/tests/test_items.py
Normal file
55
backend/tests/test_items.py
Normal file
@@ -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"
|
||||
39
backend/tests/test_warranty.py
Normal file
39
backend/tests/test_warranty.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user