"""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, joinedload, selectinload from ..database import get_db from ..deps import get_current_user, require_admin from ..models import ( Item, ItemDocument, ItemLoan, ItemUsage, Location, Movement, MovementType, Product, Shop, User, ) from ..schemas import ( DocumentSuggestions, ItemCreate, ItemDocumentUploadOut, ItemOut, ItemRemove, ItemUpdate, ZubehoerZuordnen, ) 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, ) 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 _zubehoer_freigeben(item: Item) -> None: """Zubehoer vom Geraet loesen, bevor das Geraet verschwindet. ON DELETE SET NULL steht zwar am Fremdschluessel, aber SQLite erzwingt Fremdschluessel nicht - in den Tests bliebe sonst eine tote Zuordnung stehen. Dieselbe Vorsicht wie beim Loeschen einer Gruppe. """ for teil in list(item.zubehoer): teil.parent_item_id = None item.parent_item_id = None 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: str | 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: 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() # Wie beim Entfernen eine Bewegung schreiben (siehe remove_item). Ohne # sie kannte der Verlauf nur das Verschwinden eines Einzelstuecks, nie # sein Entstehen - und das Dashboard zeigte Abgaenge ohne Zugaenge. # Fuer den Bestand aendert das nichts: der zaehlt bei Einzelstuecken die # Zeilen in items (services/stock.py), er rechnet nicht aus Bewegungen. db.add( Movement( product_id=product.id, lot_id=None, user_id=user.id, type=MovementType.in_, quantity=1, unit_used="Stück", note=(payload.note or f"Einzelstück {item.uid}"), location_id=payload.location_id, ) ) created.append(item) db.commit() for item in created: db.refresh(item) return [item_to_out(i) for i in created] @router.get("/items", response_model=list[ItemOut]) def list_all_items( db: Session = Depends(get_db), _: User = Depends(get_current_user), ) -> list[ItemOut]: """Alle Einzelstücke über alle Produkte – Grundlage der Einzelstück-Liste. Angereichert mit Produkt-, Kategorie-, Lagerort- und Shop-Angaben; die Beziehungen werden vorgeladen, damit die Liste nicht in N+1-Abfragen zerfällt. """ rows = ( db.query(Item) .options( joinedload(Item.product).joinedload(Product.category), joinedload(Item.location), joinedload(Item.shop), joinedload(Item.documents), # Verleihe samt Konto mitladen: Sonst fragt item_to_out je # verliehenem Stueck einzeln nach (N+1). joinedload(Item.loans).joinedload(ItemLoan.user), # selectin und nicht joined: Hier haengen schon drei Sammlungen # dran, eine vierte vervielfachte die Zeilen ueber Kreuz. So kostet # es eine zusaetzliche Abfrage statt eines Kreuzprodukts. selectinload(Item.usages).joinedload(ItemUsage.user), ) .order_by(Item.id) .all() ) return [item_to_out(i) for i in rows] @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) _zubehoer_freigeben(item) 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).""" item = _item_or_404(db, item_id) _zubehoer_freigeben(item) db.delete(item) db.commit() # -------------------------------------------------------------------------- # Zubehoer: ein Einzelstueck gehoert zu einem anderen (Netzteil -> Laptop) # -------------------------------------------------------------------------- def _ringpruefung(db: Session, geraet: Item, teil: Item) -> None: """Verhindert, dass ein Stueck ueber Umwege sein eigenes Zubehoer wird. Mit nur einem Elternteil je Teil ist die Kette kurz, aber A->B->A waere ohne Pruefung moeglich - und danach liefe jedes Hochlaufen der Kette endlos. Deshalb vom Geraet aus nach oben gehen, wie beim Kategoriebaum. """ if geraet.id == teil.id: raise HTTPException( status.HTTP_409_CONFLICT, "Ein Einzelstück kann nicht sein eigenes Zubehör sein" ) gesehen: set[int] = set() lauf = geraet.gehoert_zu while lauf is not None and lauf.id not in gesehen: if lauf.id == teil.id: raise HTTPException( status.HTTP_409_CONFLICT, f"„{teil.uid}“ ist bereits weiter oben in dieser Zubehör-Kette", ) gesehen.add(lauf.id) lauf = lauf.gehoert_zu @router.post("/items/{item_id}/zubehoer", response_model=ItemOut) def zubehoer_zuordnen( item_id: int, payload: ZubehoerZuordnen, db: Session = Depends(get_db), _: User = Depends(require_admin), ) -> ItemOut: """Ein Einzelstück als Zubehör an dieses Gerät hängen. Eigener Endpunkt und nicht über ``ItemUpdate``: Das Schema sendet bewusst immer alle Felder mit, auch leere, damit sich Felder loeschen lassen. Eine Beziehung darueber zu pflegen hiesse, dass ein Aufrufer sie still kappt, sobald er das Feld vergisst - genau so gingen schon einmal Kaufpreis und Waehrung verloren. """ geraet = _item_or_404(db, item_id) teil = _item_or_404(db, payload.child_item_id) _ringpruefung(db, geraet, teil) # Ein Teil haengt an genau einem Geraet - ein Wechsel ist deshalb einfach # ein Umhaengen und braucht kein vorheriges Loesen. teil.parent_item_id = geraet.id db.commit() db.refresh(geraet) return item_to_out(geraet) @router.delete( "/items/{item_id}/zubehoer/{child_item_id}", status_code=status.HTTP_204_NO_CONTENT ) def zubehoer_loesen( item_id: int, child_item_id: int, db: Session = Depends(get_db), _: User = Depends(require_admin), ) -> None: """Zubehör vom Gerät lösen. Das Stück selbst bleibt bestehen.""" geraet = _item_or_404(db, item_id) teil = _item_or_404(db, child_item_id) if teil.parent_item_id != geraet.id: raise HTTPException( status.HTTP_404_NOT_FOUND, f"„{teil.uid}“ gehört nicht zu „{geraet.uid}“", ) teil.parent_item_id = None 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" 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: 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." ) 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, filename=_safe_filename(file.filename), content_type=content_type, data=data, ) db.add(doc) db.commit() db.refresh(doc) 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, **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, 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()