91adee5333
Neuer Endpunkt ordnet OCR-Zeilen den aehnlichsten LEBENSMITTELN zu (Score 0-100, difflib + Token-/Praefix-Abgleich fuer abgekuerzte Kassennamen). Gegenstaende inkl. Verbrauchsgegenstaende bleiben aussen vor. Je Zeile die besten Treffer ueber dem Schwellwert (Request oder Einstellung receipt_match_threshold, Default 45). 3 Tests, gesamt 182 gruen. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
614 lines
22 KiB
Python
614 lines
22 KiB
Python
import difflib
|
||
import re
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, Depends, File, HTTPException, UploadFile, status
|
||
from fastapi.responses import Response
|
||
from sqlalchemy.orm import Session, joinedload, selectinload
|
||
|
||
from ..crud import product_to_out, product_tracking, products_to_out_bulk
|
||
from ..database import SessionLocal, get_db
|
||
from ..deps import get_current_user, require_admin
|
||
from ..models import (
|
||
Barcode,
|
||
BaseUnit,
|
||
Category,
|
||
CategoryTracking,
|
||
Group,
|
||
Location,
|
||
Movement,
|
||
MovementType,
|
||
Product,
|
||
ProductImage,
|
||
ProductLocationMinStock,
|
||
RemovalReason,
|
||
Shop,
|
||
User,
|
||
)
|
||
from ..off import lookup_barcode
|
||
from ..schemas import (
|
||
BarcodeCreate,
|
||
LocationMinStockIn,
|
||
LookupResult,
|
||
MatchCandidate,
|
||
MatchLine,
|
||
MatchRequest,
|
||
ProductCreate,
|
||
ProductOut,
|
||
ProductUpdate,
|
||
RemovalHistoryItem,
|
||
RemovalStat,
|
||
RemovalSummary,
|
||
)
|
||
from ..services import images
|
||
from ..services.categories import suggest_category
|
||
from .categories import descendant_ids
|
||
from .settings import get_receipt_match_threshold
|
||
from ..services.conversion import ConversionError, resolve_product_unit
|
||
from ..services.fields import FieldError, apply_field_values
|
||
from ..services.group_codes import detach as detach_group_code, sync as sync_group_code
|
||
from ..services.stock import removal_stats
|
||
|
||
router = APIRouter(prefix="/products", tags=["products"])
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Kassenzettel-Abgleich: eine OCR-Zeile den ähnlichsten Lebensmitteln zuordnen.
|
||
# ---------------------------------------------------------------------------
|
||
_UMLAUTE = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss"})
|
||
|
||
|
||
def _normalisieren(text: str) -> str:
|
||
"""Klein, Umlaute aufgelöst, nur Buchstaben/Ziffern – Preise/Sonderzeichen weg."""
|
||
text = (text or "").lower().translate(_UMLAUTE)
|
||
return re.sub(r"[^a-z0-9]+", " ", text).strip()
|
||
|
||
|
||
def _match_score(zeile: str, name: str) -> int:
|
||
"""Ähnlichkeit 0–100 zwischen Kassenzeile und Produktname.
|
||
|
||
Kombiniert die Gesamt-Ähnlichkeit mit einem Token-/Präfix-Abgleich, damit
|
||
abgekürzte Kassennamen ("MÜHLEN SCHNITZ") auf den vollen Namen passen.
|
||
"""
|
||
a, b = _normalisieren(zeile), _normalisieren(name)
|
||
if not a or not b:
|
||
return 0
|
||
gesamt = difflib.SequenceMatcher(None, a, b).ratio()
|
||
a_tok, b_tok = a.split(), b.split()
|
||
bester = []
|
||
for t in a_tok:
|
||
m = 0.0
|
||
for u in b_tok:
|
||
if u.startswith(t) or t.startswith(u):
|
||
m = max(m, min(len(t), len(u)) / max(len(t), len(u)))
|
||
else:
|
||
m = max(m, difflib.SequenceMatcher(None, t, u).ratio())
|
||
bester.append(m)
|
||
token = sum(bester) / len(bester) if bester else 0.0
|
||
return round(100 * max(gesamt, token))
|
||
|
||
|
||
@router.post("/match", response_model=list[MatchLine])
|
||
def match_receipt(
|
||
payload: MatchRequest,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> list[MatchLine]:
|
||
"""Kassenzettel-Zeilen den ähnlichsten LEBENSMITTELN zuordnen (Score 0–100).
|
||
Gegenstände (auch Verbrauchsgegenstände) bleiben außen vor; je Zeile die besten
|
||
Treffer über dem Schwellwert (aus Request oder Einstellung)."""
|
||
schwelle = payload.threshold if payload.threshold is not None else get_receipt_match_threshold(db)
|
||
schwelle = max(0, min(100, schwelle))
|
||
lebensmittel = [
|
||
p
|
||
for p in db.query(Product).options(joinedload(Product.category)).all()
|
||
if product_tracking(db, p) != CategoryTracking.object.value
|
||
]
|
||
ergebnis: list[MatchLine] = []
|
||
for zeile in payload.lines:
|
||
text = (zeile or "").strip()
|
||
treffer = [(p, _match_score(text, p.name)) for p in lebensmittel]
|
||
treffer = sorted(
|
||
(ps for ps in treffer if ps[1] >= schwelle),
|
||
key=lambda ps: ps[1],
|
||
reverse=True,
|
||
)
|
||
ergebnis.append(MatchLine(
|
||
text=text,
|
||
candidates=[
|
||
MatchCandidate(product_id=p.id, name=p.name, brand=p.brand, score=s)
|
||
for p, s in treffer[:5]
|
||
],
|
||
))
|
||
return ergebnis
|
||
|
||
|
||
@router.get("", response_model=list[ProductOut])
|
||
def list_products(
|
||
q: str | None = None,
|
||
category_id: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> list[ProductOut]:
|
||
"""Artikel auflisten, optional nach Name und Kategorie eingegrenzt.
|
||
|
||
``category_id=0`` liefert die Artikel ohne Kategorie. Bei einer echten
|
||
Kategorie zählen die Unterkategorien mit: Wer auf "Süßwaren" filtert, will
|
||
auch "Schokolade" sehen.
|
||
"""
|
||
query = db.query(Product).options(
|
||
# Relationen vorab laden, damit products_to_out_bulk kein N+1 auslöst.
|
||
joinedload(Product.category),
|
||
joinedload(Product.shop),
|
||
joinedload(Product.display_unit),
|
||
joinedload(Product.min_stock_unit),
|
||
selectinload(Product.field_values),
|
||
selectinload(Product.location_min_stocks).joinedload(ProductLocationMinStock.location),
|
||
)
|
||
if q:
|
||
like = f"%{q}%"
|
||
query = query.filter(Product.name.ilike(like))
|
||
if category_id == 0:
|
||
query = query.filter(Product.category_id.is_(None))
|
||
elif category_id is not None:
|
||
query = query.filter(Product.category_id.in_(descendant_ids(db, category_id)))
|
||
products = query.order_by(Product.name).all()
|
||
return products_to_out_bulk(db, products)
|
||
|
||
|
||
@router.get("/lookup", response_model=LookupResult)
|
||
def lookup(
|
||
barcode: str,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> LookupResult:
|
||
"""Barcode auflösen: Produkt (Haupt- oder Alias-Code), Gruppen-Code, sonst OFF."""
|
||
code = barcode.strip()
|
||
|
||
existing = db.query(Product).filter(Product.barcode == code).first()
|
||
if existing is None:
|
||
alias = db.query(Barcode).filter(Barcode.code == code).first()
|
||
if alias is not None and alias.product_id:
|
||
existing = db.get(Product, alias.product_id)
|
||
if existing:
|
||
return LookupResult(found=True, existing_product=product_to_out(db, existing))
|
||
|
||
# Code kann einer Gruppe zugeordnet sein (z.B. alle Mehl-Marken in "Mehl").
|
||
group_id = group_name = None
|
||
alias = db.query(Barcode).filter(Barcode.code == code).first()
|
||
if alias is not None and alias.group_id:
|
||
group = db.get(Group, alias.group_id)
|
||
if group is not None:
|
||
group_id, group_name = group.id, group.name
|
||
|
||
suggestion = lookup_barcode(code)
|
||
# Die OFF-Einordnung steuert die KATEGORIE (Ueberblick), nicht die Gruppe.
|
||
category = suggest_category(db, suggestion)
|
||
category_id = category.id if category else None
|
||
category_name = category.name if category else None
|
||
if suggestion:
|
||
return LookupResult(
|
||
found=True, suggestion=suggestion, group_id=group_id, group_name=group_name,
|
||
category_id=category_id, category_name=category_name,
|
||
)
|
||
return LookupResult(found=False, group_id=group_id, group_name=group_name)
|
||
|
||
|
||
@router.get("/{product_id}", response_model=ProductOut)
|
||
def get_product(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> ProductOut:
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
return product_to_out(db, product)
|
||
|
||
|
||
@router.put("/{product_id}/location-min-stock", response_model=ProductOut)
|
||
def set_product_location_min_stock(
|
||
product_id: int,
|
||
payload: list[LocationMinStockIn],
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> ProductOut:
|
||
"""Mindestbestände je Lagerort komplett ersetzen (Menge 0 = Eintrag entfällt)."""
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
|
||
db.query(ProductLocationMinStock).filter(
|
||
ProductLocationMinStock.product_id == product_id
|
||
).delete()
|
||
gesehen: set[int] = set()
|
||
for eintrag in payload:
|
||
if eintrag.min_stock <= 0 or eintrag.location_id in gesehen:
|
||
continue
|
||
if db.get(Location, eintrag.location_id) is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
|
||
gesehen.add(eintrag.location_id)
|
||
db.add(ProductLocationMinStock(
|
||
product_id=product_id,
|
||
location_id=eintrag.location_id,
|
||
min_stock=eintrag.min_stock,
|
||
))
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product_to_out(db, product)
|
||
|
||
|
||
@router.get("/{product_id}/removals", response_model=RemovalSummary)
|
||
def product_removals(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> RemovalSummary:
|
||
"""Entnahmen mit Grund: Summe je Grund plus die jüngsten Einträge."""
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
|
||
stats = [
|
||
RemovalStat(reason=RemovalReason(grund), quantity=v["quantity"], count=v["count"])
|
||
for grund, v in removal_stats(db, product_id).items()
|
||
]
|
||
stats.sort(key=lambda s: s.quantity, reverse=True)
|
||
|
||
rows = (
|
||
db.query(Movement)
|
||
.filter(
|
||
Movement.product_id == product_id,
|
||
Movement.type == MovementType.out,
|
||
Movement.reason.isnot(None),
|
||
)
|
||
.order_by(Movement.created_at.desc())
|
||
.limit(50)
|
||
.all()
|
||
)
|
||
loc_names = {loc.id: loc.name for loc in db.query(Location).all()}
|
||
user_ids = {m.user_id for m in rows if m.user_id}
|
||
users = (
|
||
{u.id: u.username for u in db.query(User).filter(User.id.in_(user_ids)).all()}
|
||
if user_ids
|
||
else {}
|
||
)
|
||
history = [
|
||
RemovalHistoryItem(
|
||
reason=RemovalReason(m.reason),
|
||
quantity=m.quantity,
|
||
location_id=m.location_id,
|
||
location_name=loc_names.get(m.location_id),
|
||
note=m.note,
|
||
username=users.get(m.user_id),
|
||
created_at=m.created_at,
|
||
)
|
||
for m in rows
|
||
]
|
||
return RemovalSummary(stats=stats, history=history)
|
||
|
||
|
||
@router.get("/{product_id}/off", response_model=LookupResult)
|
||
def off_vergleich(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> LookupResult:
|
||
"""Open Food Facts erneut befragen – auch für einen bereits angelegten Artikel.
|
||
|
||
``/lookup`` kann das nicht: Kennt es den Barcode schon, meldet es den eigenen
|
||
Artikel und fragt OFF gar nicht erst. Zum Vergleichen braucht es aber genau
|
||
die fremden Daten.
|
||
|
||
Neben dem Haupt-Barcode werden die zusätzlichen EAN-Codes durchprobiert –
|
||
oft ist nur einer davon bei OFF hinterlegt.
|
||
"""
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
|
||
codes = [product.barcode] if product.barcode else []
|
||
codes += [
|
||
b.code
|
||
for b in db.query(Barcode).filter(Barcode.product_id == product_id).all()
|
||
if b.code not in codes
|
||
]
|
||
if not codes:
|
||
raise HTTPException(
|
||
status.HTTP_400_BAD_REQUEST,
|
||
"Dieser Artikel hat keinen Barcode – ohne den kann Open Food Facts nichts finden.",
|
||
)
|
||
|
||
# Gegenstände zuerst in der allgemeinen Produkt-DB nachschlagen, Lebensmittel in OFF.
|
||
prefer = "object" if product_tracking(db, product) == "object" else None
|
||
for code in codes:
|
||
suggestion = lookup_barcode(code, prefer=prefer)
|
||
if suggestion:
|
||
category = suggest_category(db, suggestion)
|
||
return LookupResult(
|
||
found=True,
|
||
suggestion=suggestion,
|
||
category_id=category.id if category else None,
|
||
category_name=category.name if category else None,
|
||
)
|
||
return LookupResult(found=False)
|
||
|
||
|
||
@router.get("/{product_id}/image")
|
||
def get_product_image(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> Response:
|
||
"""Artikelbild aus der eigenen Datenbank – nie von Open Food Facts.
|
||
|
||
Anders als das Logo verlangt diese Route eine Anmeldung: Aus den Bildern
|
||
liesse sich sonst ohne Konto ablesen, was im Vorrat liegt.
|
||
"""
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
|
||
bild = images.ensure(db, product)
|
||
if bild is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Kein Bild vorhanden")
|
||
return Response(
|
||
content=bild.data,
|
||
media_type=bild.content_type,
|
||
headers={
|
||
"Cache-Control": "private, max-age=300",
|
||
"ETag": f'"bild-{product_id}-{int(bild.updated_at.timestamp())}"',
|
||
},
|
||
)
|
||
|
||
|
||
@router.put("/{product_id}/image", response_model=ProductOut)
|
||
async def upload_product_image(
|
||
product_id: int,
|
||
file: UploadFile = File(...),
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> ProductOut:
|
||
"""Eigenes Foto hochladen (Kamera oder Galerie) und beim Artikel ablegen.
|
||
|
||
``source_url = None`` markiert das Bild als selbst hochgeladen – so ersetzt
|
||
es der Bildabgleich beim Speichern nicht durch ein Bild aus einer Adresse
|
||
(solange keine Bildadresse gesetzt ist).
|
||
"""
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
if file.content_type not in images.ALLOWED_TYPES:
|
||
erlaubt = ", ".join(sorted(t.split("/")[-1] for t in images.ALLOWED_TYPES))
|
||
raise HTTPException(
|
||
status.HTTP_400_BAD_REQUEST,
|
||
f"Dieses Bildformat wird nicht unterstützt. Erlaubt sind: {erlaubt}.",
|
||
)
|
||
data = await file.read()
|
||
if not data:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Die Datei ist leer.")
|
||
if len(data) > images.MAX_BYTES:
|
||
raise HTTPException(
|
||
status.HTTP_400_BAD_REQUEST,
|
||
f"Das Bild ist zu groß ({len(data) // 1024} KB). "
|
||
f"Erlaubt sind höchstens {images.MAX_BYTES // 1024} KB.",
|
||
)
|
||
bild = db.get(ProductImage, product.id)
|
||
if bild is None:
|
||
bild = ProductImage(
|
||
product_id=product.id, content_type=file.content_type, data=data, source_url=None
|
||
)
|
||
db.add(bild)
|
||
else:
|
||
bild.content_type, bild.data, bild.source_url = file.content_type, data, None
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product_to_out(db, product)
|
||
|
||
|
||
@router.delete("/{product_id}/image", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_product_image(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> None:
|
||
"""Foto entfernen. Eine gesetzte Bildadresse bleibt bestehen."""
|
||
bild = db.get(ProductImage, product_id)
|
||
if bild is not None:
|
||
db.delete(bild)
|
||
db.commit()
|
||
|
||
|
||
def _store_product_image_bg(product_id: int, url: str) -> None:
|
||
"""Bild nach dem Anlegen im Hintergrund holen (eigene Session), damit das
|
||
Anlegen nicht auf den (langsamen) Netzwerk-Abruf wartet."""
|
||
db = SessionLocal()
|
||
try:
|
||
product = db.get(Product, product_id)
|
||
if product is not None:
|
||
images.store(db, product, url)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@router.post("", response_model=ProductOut, status_code=status.HTTP_201_CREATED)
|
||
def create_product(
|
||
payload: ProductCreate,
|
||
background_tasks: BackgroundTasks,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> ProductOut:
|
||
if payload.barcode:
|
||
exists = db.query(Product).filter(Product.barcode == payload.barcode).first()
|
||
if exists:
|
||
raise HTTPException(
|
||
status.HTTP_409_CONFLICT, "Ein Produkt mit diesem Barcode existiert bereits"
|
||
)
|
||
base_unit = payload.base_unit
|
||
display_unit_id = None
|
||
if payload.unit_id is not None:
|
||
try:
|
||
base_unit, display_unit_id = resolve_product_unit(db, payload.unit_id)
|
||
except ConversionError as exc:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
||
if payload.shop_id is not None and db.get(Shop, payload.shop_id) is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Shop nicht gefunden")
|
||
product = Product(
|
||
barcode=payload.barcode or None,
|
||
name=payload.name,
|
||
brand=payload.brand,
|
||
image_url=payload.image_url,
|
||
base_unit=base_unit,
|
||
display_unit_id=display_unit_id,
|
||
package_size=payload.package_size,
|
||
package_label=payload.package_label or None,
|
||
date_precision=payload.date_precision.value,
|
||
group_id=payload.group_id,
|
||
category_id=payload.category_id,
|
||
min_stock=payload.min_stock,
|
||
min_stock_unit_id=payload.min_stock_unit_id,
|
||
min_stock_in_packages=bool(payload.min_stock_in_packages),
|
||
shop_id=payload.shop_id,
|
||
product_url=payload.product_url or None,
|
||
individual=bool(payload.individual),
|
||
bulk=bool(payload.bulk),
|
||
source="manual",
|
||
)
|
||
db.add(product)
|
||
db.flush() # product.id fuer den Gruppen-Code
|
||
sync_group_code(db, product)
|
||
try:
|
||
apply_field_values(db, product, payload.field_values)
|
||
except FieldError as exc:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
||
db.commit()
|
||
db.refresh(product)
|
||
if product.image_url:
|
||
# Bild erst NACH dem Anlegen im Hintergrund holen – der Netzwerk-Abruf
|
||
# soll das Anlegen nicht mehrere Sekunden blockieren. Schlägt er fehl,
|
||
# bleibt der Artikel trotzdem angelegt (nur ohne Bild).
|
||
background_tasks.add_task(_store_product_image_bg, product.id, product.image_url)
|
||
return product_to_out(db, product)
|
||
|
||
|
||
@router.patch("/{product_id}", response_model=ProductOut)
|
||
def update_product(
|
||
product_id: int,
|
||
payload: ProductUpdate,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> ProductOut:
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
|
||
data = payload.model_dump(exclude_unset=True)
|
||
# Feldwerte sind keine Spalte, sondern eigene Zeilen – getrennt behandeln.
|
||
field_values = data.pop("field_values", None)
|
||
if "barcode" in data and data["barcode"]:
|
||
clash = (
|
||
db.query(Product)
|
||
.filter(Product.barcode == data["barcode"], Product.id != product_id)
|
||
.first()
|
||
)
|
||
if clash:
|
||
raise HTTPException(
|
||
status.HTTP_409_CONFLICT, "Ein anderes Produkt hat diesen Barcode bereits"
|
||
)
|
||
if data.get("shop_id") is not None and db.get(Shop, data["shop_id"]) is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Shop nicht gefunden")
|
||
# Einheit: unit_id (falls gesetzt) bestimmt base_unit + Anzeigeeinheit.
|
||
if "unit_id" in data:
|
||
unit_id = data.pop("unit_id")
|
||
data.pop("base_unit", None) # unit_id hat Vorrang
|
||
if unit_id is None:
|
||
product.display_unit_id = None
|
||
else:
|
||
try:
|
||
product.base_unit, product.display_unit_id = resolve_product_unit(db, unit_id)
|
||
except ConversionError as exc:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
||
if data.get("min_stock_in_packages") is None:
|
||
data.pop("min_stock_in_packages", None) # Spalte ist NOT NULL
|
||
if data.get("date_precision") is None:
|
||
data.pop("date_precision", None) # Spalte ist NOT NULL
|
||
else:
|
||
data["date_precision"] = data["date_precision"].value
|
||
for field, value in data.items():
|
||
setattr(product, field, value)
|
||
if field_values is not None:
|
||
try:
|
||
apply_field_values(db, product, field_values)
|
||
except FieldError as exc:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
||
# Gruppe oder Barcode koennen sich geaendert haben - Code nachziehen.
|
||
sync_group_code(db, product)
|
||
if "image_url" in data:
|
||
# Massgeblich ist, woher die vorhandene Kopie stammt - nicht, ob sich
|
||
# die Adresse am Artikel geaendert hat. Sonst passierte genau dann
|
||
# nichts, wenn die Adresse schon stimmte, der Abruf damals aber
|
||
# fehlschlug: Der Artikel bliebe dauerhaft ohne Bild.
|
||
alt = db.get(ProductImage, product.id)
|
||
if alt is not None and alt.source_url != product.image_url:
|
||
db.delete(alt)
|
||
db.flush()
|
||
alt = None
|
||
if alt is None and product.image_url:
|
||
images.store(db, product, product.image_url)
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product_to_out(db, product)
|
||
|
||
|
||
@router.post("/{product_id}/barcodes", response_model=ProductOut, status_code=status.HTTP_201_CREATED)
|
||
def add_product_barcode(
|
||
product_id: int,
|
||
payload: BarcodeCreate,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> ProductOut:
|
||
"""Weiteren EAN-Code zu einem Produkt hinzufügen."""
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
code = payload.code.strip()
|
||
if db.query(Barcode).filter(Barcode.code == code).first() or (
|
||
db.query(Product).filter(Product.barcode == code).first()
|
||
):
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "Dieser Code ist bereits vergeben")
|
||
db.add(Barcode(code=code, note=(payload.note or None), product_id=product.id))
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product_to_out(db, product)
|
||
|
||
|
||
@router.delete("/{product_id}/barcodes/{code}", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_product_barcode(
|
||
product_id: int,
|
||
code: str,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> None:
|
||
entry = (
|
||
db.query(Barcode)
|
||
.filter(Barcode.product_id == product_id, Barcode.code == code)
|
||
.first()
|
||
)
|
||
if entry is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Code nicht gefunden")
|
||
db.delete(entry)
|
||
db.commit()
|
||
|
||
|
||
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_product(
|
||
product_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> None:
|
||
product = db.get(Product, product_id)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
detach_group_code(db, product)
|
||
db.delete(product)
|
||
db.commit()
|