"""Artikelbilder holen und lokal vorhalten. Open Food Facts liefert nur eine Bild-*Adresse*. Würde die Oberfläche direkt dorthin verlinken, hinge jede Artikelseite an einem fremden Dienst: Das Bild verschwindet, wenn OFF es austauscht oder nicht erreichbar ist, die Installation funktioniert nicht mehr offline, und jeder Seitenaufruf verrät OFF, welche Artikel jemand ansieht. Deshalb wird das Bild einmal geholt und danach aus der eigenen Datenbank ausgeliefert. Die Bilder liegen in einer eigenen Tabelle und nicht als Spalte an ``products``: Sonst zöge jede Artikelliste die Blobs mit. """ from __future__ import annotations import httpx from sqlalchemy.orm import Session from ..models import Product, ProductImage # Grosszuegig genug fuer ein Produktfoto (auch ein Handy-Foto), eng genug, dass # niemand die Datenbank mit einer versehentlich verlinkten Datei volllaeuft. # Bleibt unter dem nginx-Limit (client_max_body_size, siehe web/nginx.conf), # damit Uebergroesse eine klare Meldung statt eines nackten 413 ergibt. MAX_BYTES = 8 * 1024 * 1024 ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"} TIMEOUT_SECONDS = 15.0 def fetch(url: str) -> tuple[bytes, str] | None: """Bild herunterladen. Gibt (Daten, Inhaltstyp) zurück oder None. Fehler sind hier bewusst kein Ausnahmefall, der nach oben durchschlägt: Ein fehlendes Bild darf niemals das Anlegen eines Artikels verhindern. """ if not url or not url.startswith(("http://", "https://")): return None try: resp = httpx.get( url, timeout=TIMEOUT_SECONDS, follow_redirects=True, headers={"User-Agent": "Vorrania-Selfhosted/1.0"}, ) except httpx.HTTPError: return None if resp.status_code != 200: return None typ = (resp.headers.get("content-type") or "").split(";")[0].strip().lower() if typ not in ALLOWED_TYPES: return None daten = resp.content if not daten or len(daten) > MAX_BYTES: return None return daten, typ def store(db: Session, product: Product, url: str) -> ProductImage | None: """Bild holen und beim Artikel ablegen. Ohne Erfolg bleibt alles wie es war.""" geholt = fetch(url) if geholt is None: return None daten, typ = geholt bild = db.get(ProductImage, product.id) if bild is None: bild = ProductImage( product_id=product.id, content_type=typ, data=daten, source_url=url ) db.add(bild) else: bild.content_type, bild.data, bild.source_url = typ, daten, url return bild def ensure(db: Session, product: Product) -> ProductImage | None: """Lokale Kopie zurückgeben und bei Bedarf einmalig nachholen. So bekommen auch Artikel ein lokales Bild, die vor dieser Funktion angelegt wurden – ohne Wanderung über alle Datensätze. Bezahlt wird das mit einer einmaligen Verzögerung beim ersten Aufruf. """ bild = db.get(ProductImage, product.id) if bild is not None: return bild if not product.image_url: return None bild = store(db, product, product.image_url) if bild is not None: db.commit() return bild