2bc871f229
Findet die Beleg-Analyse mehrere Beträge, liefert der Upload jetzt eine Kandidatenliste (bester Tipp zuerst). Web und iOS bieten dann ein Dropdown zur Auswahl des richtigen Kaufpreises, statt nur den automatischen Tipp. 1 neuer Test; Backend-Suite gruen. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
295 lines
9.0 KiB
Python
295 lines
9.0 KiB
Python
"""Einzelstücke (Items) eines Gegenstands: anlegen, ändern, entfernen, per UID finden.
|
||
|
||
Ein Item ist ein physisches Exemplar mit eigener UID/QR und eigenen Angaben
|
||
(Lagerort, Kaufdatum, Garantie, Bezugsquelle, Notiz). Nur sinnvoll für
|
||
Gegenstands-Produkte mit ``Product.individual``.
|
||
"""
|
||
|
||
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,
|
||
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_price_candidates,
|
||
guess_price_cents,
|
||
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)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
return product
|
||
|
||
|
||
def _item_or_404(db: Session, item_id: int) -> Item:
|
||
item = db.get(Item, item_id)
|
||
if item is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Einzelstück nicht gefunden")
|
||
return item
|
||
|
||
|
||
def _check_refs(db: Session, shop_id: int | None, location_id: int | None) -> None:
|
||
if shop_id is not None and db.get(Shop, shop_id) is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Shop nicht gefunden")
|
||
if location_id is not None and db.get(Location, location_id) is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
|
||
|
||
|
||
@router.get("/products/{product_id}/items", response_model=list[ItemOut])
|
||
def list_items(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> list[ItemOut]:
|
||
_product_or_404(db, product_id)
|
||
rows = db.query(Item).filter(Item.product_id == product_id).order_by(Item.id).all()
|
||
return [item_to_out(i) for i in rows]
|
||
|
||
|
||
@router.post(
|
||
"/products/{product_id}/items", response_model=list[ItemOut],
|
||
status_code=status.HTTP_201_CREATED,
|
||
)
|
||
def create_items(
|
||
product_id: int,
|
||
payload: ItemCreate,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> list[ItemOut]:
|
||
"""Ein oder mehrere (count) Einzelstücke mit gemeinsamen Startwerten anlegen."""
|
||
product = _product_or_404(db, product_id)
|
||
_check_refs(db, payload.shop_id, payload.location_id)
|
||
created: list[Item] = []
|
||
for _ in range(payload.count):
|
||
item = Item(
|
||
uid=generate_uid(db),
|
||
product_id=product.id,
|
||
location_id=payload.location_id,
|
||
shop_id=payload.shop_id,
|
||
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()
|
||
created.append(item)
|
||
db.commit()
|
||
for item in created:
|
||
db.refresh(item)
|
||
return [item_to_out(i) for i in created]
|
||
|
||
|
||
@router.get("/items/by-uid/{uid}", response_model=ItemOut)
|
||
def item_by_uid(
|
||
uid: str,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> ItemOut:
|
||
"""QR-Auflösung: UID → Einzelstück (mit Produktangaben)."""
|
||
item = db.query(Item).filter(Item.uid == uid.strip().upper()).first()
|
||
if item is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Kein Einzelstück mit dieser UID")
|
||
return item_to_out(item)
|
||
|
||
|
||
@router.get("/items/{item_id}", response_model=ItemOut)
|
||
def get_item(
|
||
item_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> ItemOut:
|
||
return item_to_out(_item_or_404(db, item_id))
|
||
|
||
|
||
@router.patch("/items/{item_id}", response_model=ItemOut)
|
||
def update_item(
|
||
item_id: int,
|
||
payload: ItemUpdate,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> ItemOut:
|
||
item = _item_or_404(db, item_id)
|
||
data = payload.model_dump(exclude_unset=True)
|
||
_check_refs(db, data.get("shop_id"), data.get("location_id"))
|
||
for field, value in data.items():
|
||
setattr(item, field, value)
|
||
db.commit()
|
||
db.refresh(item)
|
||
return item_to_out(item)
|
||
|
||
|
||
@router.post("/items/{item_id}/remove", status_code=status.HTTP_204_NO_CONTENT)
|
||
def remove_item(
|
||
item_id: int,
|
||
payload: ItemRemove,
|
||
db: Session = Depends(get_db),
|
||
user: User = Depends(require_admin),
|
||
) -> None:
|
||
"""Einzelstück mit Grund entfernen – als Bewegung protokolliert (Statistik je Modell)."""
|
||
item = _item_or_404(db, item_id)
|
||
db.add(
|
||
Movement(
|
||
product_id=item.product_id,
|
||
lot_id=None,
|
||
user_id=user.id,
|
||
type=MovementType.out,
|
||
quantity=1,
|
||
unit_used="Stück",
|
||
note=(payload.note or f"Einzelstück {item.uid}"),
|
||
location_id=item.location_id,
|
||
reason=payload.reason.value,
|
||
)
|
||
)
|
||
db.delete(item)
|
||
db.commit()
|
||
|
||
|
||
@router.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_item(
|
||
item_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> None:
|
||
"""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 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)
|
||
|
||
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,
|
||
)
|
||
|
||
|
||
@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()
|