diff --git a/.env.example b/.env.example index 60ed0d3..a0d0155 100644 --- a/.env.example +++ b/.env.example @@ -43,13 +43,28 @@ OIDC_SCOPES=openid profile email # --- Logo-Service ------------------------------------------------------------ # Verzeichnis für den lokalen Logo-Cache (im Container auf das Volume gemountet). +# Logos werden ausschließlich von hier ausgeliefert – im Seitenaufruf entsteht +# kein Zugriff auf fremde Dienste. LOGO_STORAGE_DIR=/data/logos -# Optionale Provider. Ohne Key wird der jeweilige Adapter übersprungen. + +# Der lokale simple-icons-Katalog deckt die meisten Marken ohne Netzzugriff ab. +# Die folgenden Provider sind optional; ohne Schlüssel wird der jeweilige +# Adapter stillschweigend übersprungen. + +# logo.dev: veröffentlichbarer Token von https://www.logo.dev +# Achtung: Auf dem kostenlosen Tarif verlangt logo.dev bei kommerzieller +# Nutzung einen sichtbaren Rückverweis. Für private Nutzung entfällt das. LOGODEV_API_KEY= + +# Brandfetch: API-Schlüssel der Brand API (https://docs.brandfetch.com). +# Wird nur genutzt, wenn zur Firma eine Domain hinterlegt ist. BRANDFETCH_API_KEY= + LOGO_HTTP_TIMEOUT_SECONDS=5 LOGO_HTTP_RETRIES=2 LOGO_MAX_UPLOAD_BYTES=1048576 +# Sucht nach dem Anlegen einer Firma automatisch im Hintergrund nach einem Logo. +LOGO_AUTO_RESOLVE=true # --- Benachrichtigungen ------------------------------------------------------ NOTIFICATIONS_ENABLED=true diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a33267..150f823 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,7 +51,24 @@ die Versionierung folgt [Semantic Versioning](https://semver.org/lang/de/). Rücklagen und Vergleich zum Vormonat. - Vollständig annotiertes OpenAPI-Dokument unter `/api/docs`. +- Logo-Service mit fünfstufiger Provider-Kette: lokal gevendorte simple-icons + (3.459 Marken, rund 2 MB), logo.dev und Brandfetch bei gesetztem Schlüssel, + Favicon-Fallback und generierter Buchstaben-Avatar als Garantie. +- Logo-Cache im Dateisystem, benannt nach dem SHA-256 des Inhalts; Auslieferung + ausschließlich über `GET /api/logos/{id}` mit `immutable`-Cache-Header und + ETag-Unterstützung. +- Ermittlung der Markenfarbe: bei SVG aus den `fill`-Attributen, bei + Rastergrafiken per k-Means über 64×64 Pixel; dazu eine im HSL-Raum aufgehellte + Variante, die gegen den dunklen Hintergrund mindestens 4,5:1 Kontrast erreicht. +- Auswahldialog-Endpunkte: Kandidatensuche mit Vorauswahl, Übernahme eines + Kandidaten, Upload (SVG/PNG/JPEG bis 1 MB) und Zurücksetzen. +- Die Logosuche läuft nach dem Anlegen einer Firma im Hintergrund; der Aufruf + antwortet sofort. +- `make vendor-icons` erzeugt den simple-icons-Index neu. + ### Geändert - `SECRET_KEY` muss mindestens 32 Zeichen lang sein (Vorgabe von HS256); in Produktion wird der Platzhalterwert beim Start abgelehnt. +- Der Bildtyp eines Uploads wird ausschließlich am Inhalt bestimmt; die vom + Aufrufer gemeldete Kopfzeile wird nicht mehr als Nachweis akzeptiert. diff --git a/Makefile b/Makefile index 33469d2..176d9eb 100644 --- a/Makefile +++ b/Makefile @@ -27,6 +27,10 @@ revision: ## Neue Migration erzeugen: make revision m="beschreibung" seed: ## Stammdaten (Kategoriebaum, Benachrichtigungsregeln) anlegen cd $(BACKEND) && .venv/bin/python -m app.scripts.seed +.PHONY: vendor-icons +vendor-icons: ## simple-icons-Index neu erzeugen (benötigt npm und Netzzugang) + cd $(BACKEND) && .venv/bin/python scripts/vendor_simple_icons.py + .PHONY: dev dev: ## Backend mit Hot Reload starten cd $(BACKEND) && .venv/bin/uvicorn app.main:app --reload --port 8000 diff --git a/README.md b/README.md index bb9697e..dad0e57 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ _Platzhalter – werden ergänzt, sobald die Oberfläche steht._ - Verträge mit Mindestlaufzeit, Kündigungsfrist und automatischer Verlängerung. - Ratenzahlungen mit Restschuld- und Restratenberechnung. - Rücklagenbildung für nicht-monatliche Posten. -- Firmenlogos und Markenfarben, lokal zwischengespeichert. +- Firmenlogos und Markenfarben, lokal zwischengespeichert (siehe unten). - Auswertungen: Monatsübersicht, Cashflow-Kalender, 12-Monats-Forecast, Kategorien, Abo-Übersicht, Jahresvergleich, Budgets, Sparziele. - Benachrichtigungen per SMTP und Apprise. @@ -69,12 +69,37 @@ vollständige, kommentierte Liste. Die wichtigsten: | `TIMEZONE` | `Europe/Berlin` | Zeitzone der gesamten Anwendung | | `HOLIDAY_REGION` | `DE-NW` | Feiertagsregion für Werktagsverschiebungen | | `LOGO_STORAGE_DIR` | `/data/logos` | Verzeichnis des Logo-Caches | -| `LOGODEV_API_KEY` / `BRANDFETCH_API_KEY` | leer | Optionale Logo-Provider | +| `LOGODEV_API_KEY` / `BRANDFETCH_API_KEY` | leer | Optionale Logo-Provider, ohne Schlüssel übersprungen | | `SMTP_*` | leer | Mailversand für Benachrichtigungen | | `APPRISE_URLS` | leer | Komma-separierte Apprise-Ziele | | `COOKIE_SECURE` | `true` | Hinter HTTPS `true`, für lokales HTTP `false` | | `SCHEDULER_ENABLED` | `true` | Täglicher Benachrichtigungslauf um 07:00 | +## Logos und Markenfarben + +Jede Firma bekommt ein Bild. Die Provider-Kette arbeitet der Reihe nach: + +1. **simple-icons** – rund 3.500 Marken, im Repository unter + `backend/app/assets/simple_icons.json.gz` mitgeliefert. Kein Schlüssel, kein + Netzzugriff. Neu erzeugen mit `make vendor-icons`. +2. **logo.dev** – nur mit `LOGODEV_API_KEY`. +3. **Brandfetch** – nur mit `BRANDFETCH_API_KEY` und hinterlegter Domain. +4. **Favicon** – über den Google-Dienst, wenn eine Domain bekannt oder aus dem + Namen ableitbar ist. +5. **Generierter Avatar** – Buchstaben-Monogramm mit fester Farbe aus dem + Namens-Hash. Schlägt nie fehl. + +Findet die Kette einen eindeutigen Treffer im lokalen Katalog, unterbleiben die +Anfragen nach außen vollständig – die Firmenliste verlässt den Server nicht. Im +Auswahldialog (`POST /api/merchants/{id}/logo/search`) läuft dagegen bewusst die +ganze Kette, damit Alternativen zur Wahl stehen. + +Ausgewählte Logos landen unter `LOGO_STORAGE_DIR`, benannt nach ihrem +SHA-256-Hash, und werden nur über `GET /api/logos/{id}` ausgeliefert. Beim +Seitenaufruf entsteht dadurch kein Zugriff auf fremde Dienste. Ein Upload oder +eine bewusste Auswahl setzt den Status auf `manual` und wird von der Automatik +nie überschrieben. + ## Projektstruktur ``` diff --git a/backend/app/api/router.py b/backend/app/api/router.py index f6b17d1..ab95fd0 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -12,6 +12,7 @@ from app.api.routes import ( auth, budgets, categories, + logos, me, merchants, occurrences, @@ -34,6 +35,7 @@ protected = APIRouter(dependencies=[Depends(get_active_user)]) protected.include_router(accounts.router) protected.include_router(categories.router) protected.include_router(merchants.router) +protected.include_router(logos.router) protected.include_router(recurrences.router) protected.include_router(occurrences.router) protected.include_router(transactions.router) diff --git a/backend/app/api/routes/logos.py b/backend/app/api/routes/logos.py new file mode 100644 index 0000000..6407404 --- /dev/null +++ b/backend/app/api/routes/logos.py @@ -0,0 +1,45 @@ +"""Auslieferung der zwischengespeicherten Logos. + +Die Dateien kommen ausschließlich von der lokalen Platte; beim Seitenaufruf +entsteht kein Zugriff auf fremde Dienste. Der Dateiname ist der Inhaltshash, +daher darf unbegrenzt zwischengespeichert werden. +""" + +from fastapi import APIRouter, Request, Response, status +from fastapi.responses import FileResponse + +from app.api.deps import DbSession +from app.core.config import settings +from app.core.errors import NotFoundError +from app.models import LogoAsset +from app.schemas.common import ErrorResponse +from app.services.crud import get_or_404 +from app.services.logos import cache_headers + +router = APIRouter(prefix="/logos", tags=["logos"]) + + +@router.get( + "/{asset_id}", + responses={ + status.HTTP_200_OK: { + "content": {"image/svg+xml": {}, "image/png": {}, "image/jpeg": {}}, + "description": "Die Logodatei aus dem lokalen Cache.", + }, + status.HTTP_404_NOT_FOUND: {"model": ErrorResponse}, + }, + summary="Logo ausliefern", + description="Liefert die Datei mit `Cache-Control: public, max-age=31536000, immutable`.", +) +async def read_logo(asset_id: int, request: Request, session: DbSession) -> Response: + asset = await get_or_404(session, LogoAsset, asset_id) + pfad = settings.logo_storage_dir / asset.file_path + if not pfad.exists(): + raise NotFoundError("Die Logodatei fehlt im Cache.", code="logo_file_missing") + + headers = cache_headers(asset) + # Unveränderliche Inhalte: ein passendes ETag beantwortet die Anfrage sofort. + if request.headers.get("if-none-match") == headers["ETag"]: + return Response(status_code=status.HTTP_304_NOT_MODIFIED, headers=headers) + + return FileResponse(pfad, media_type=asset.mime, headers=headers) diff --git a/backend/app/api/routes/merchants.py b/backend/app/api/routes/merchants.py index 03f7bef..b90cfc7 100644 --- a/backend/app/api/routes/merchants.py +++ b/backend/app/api/routes/merchants.py @@ -1,16 +1,32 @@ """Firmen und Zahlungsempfänger.""" -from fastapi import APIRouter, Query, status +from fastapi import APIRouter, BackgroundTasks, File, Query, UploadFile, status from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from app.api.deps import DbSession +from app.core.config import settings from app.core.errors import ConflictError from app.models import Merchant, Recurrence, Transaction from app.models.enums import LogoStatus from app.schemas.common import ErrorResponse, MessageResponse, Page -from app.schemas.merchant import MerchantCreate, MerchantOut, MerchantUpdate +from app.schemas.merchant import ( + LogoCandidateOut, + LogoSearchOut, + LogoSelectRequest, + MerchantCreate, + MerchantOut, + MerchantUpdate, +) from app.services.crud import apply_updates, get_or_404 +from app.services.logos import ( + apply_upload, + resolve_merchant_logo, + resolve_merchant_logo_task, + search_candidates, + select_candidate, + store_candidate, +) from app.services.merchants import normalize_name, search_statement router = APIRouter(prefix="/merchants", tags=["merchants"]) @@ -48,9 +64,12 @@ async def list_merchants( response_model=MerchantOut, status_code=status.HTTP_201_CREATED, summary="Firma anlegen", - description="Antwortet sofort. Der Logo-Status steht zunächst auf `pending`.", + description="Antwortet sofort. Die Logosuche läuft anschließend im Hintergrund; " + "der Status wechselt dabei von `pending` auf `resolved`.", ) -async def create_merchant(payload: MerchantCreate, session: DbSession) -> Merchant: +async def create_merchant( + payload: MerchantCreate, background: BackgroundTasks, session: DbSession +) -> Merchant: merchant = Merchant( name=payload.name, normalized_name=normalize_name(payload.name), @@ -67,6 +86,10 @@ async def create_merchant(payload: MerchantCreate, session: DbSession) -> Mercha await session.rollback() raise ConflictError(f"Die Firma '{payload.name}' ist bereits angelegt.") from exc await session.refresh(merchant) + + if settings.logo_auto_resolve: + # Läuft erst nach der Antwort – der Aufrufer wartet nicht auf die Suche. + background.add_task(resolve_merchant_logo_task, merchant.id) return merchant @@ -116,3 +139,126 @@ async def delete_merchant(merchant_id: int, session: DbSession) -> MessageRespon await session.delete(merchant) await session.commit() return MessageResponse(detail="Firma gelöscht.") + + +# --- Logos --------------------------------------------------------------------- + + +def _to_candidate_out(asset_id: int, kandidat, mime: str, is_best: bool) -> LogoCandidateOut: + return LogoCandidateOut( + candidate_id=asset_id, + source=kandidat.source, + title=kandidat.title, + score=kandidat.score, + mime=mime, + width=kandidat.width, + height=kandidat.height, + brand_color=kandidat.brand_color, + is_preselected=is_best, + ) + + +@router.post( + "/{merchant_id}/logo/search", + response_model=LogoSearchOut, + responses=NOT_FOUND, + summary="Logos suchen", + description="Arbeitet die gesamte Provider-Kette ab und liefert bis zu fünf " + "Kandidaten. Alle werden im lokalen Cache abgelegt und sind sofort über " + "`/api/logos/{id}` abrufbar. Der beste Treffer ist vorausgewählt. Die Auswahl " + "der Firma wird dabei noch nicht verändert.", +) +async def search_logos( + merchant_id: int, + session: DbSession, + domain: str | None = Query( + default=None, description="Überschreibt die hinterlegte Domain für diese Suche." + ), +) -> LogoSearchOut: + merchant = await get_or_404(session, Merchant, merchant_id) + + if domain is not None: + merchant.domain = domain or None + + kandidaten = await search_candidates(merchant.name, merchant.domain) + ausgabe: list[LogoCandidateOut] = [] + for index, kandidat in enumerate(kandidaten): + asset = await store_candidate(session, kandidat) + ausgabe.append(_to_candidate_out(asset.id, kandidat, asset.mime, index == 0)) + + await session.commit() + return LogoSearchOut(merchant_id=merchant.id, candidates=ausgabe) + + +@router.post( + "/{merchant_id}/logo/select", + response_model=MerchantOut, + responses=NOT_FOUND, + summary="Logo auswählen", + description="Übernimmt einen Kandidaten aus der Suche. Der Status wechselt auf " + "`manual`; die automatische Suche überschreibt die Auswahl danach nicht mehr.", +) +async def select_logo(merchant_id: int, payload: LogoSelectRequest, session: DbSession) -> Merchant: + merchant = await get_or_404(session, Merchant, merchant_id) + await select_candidate(session, merchant, payload.candidate_id) + await session.commit() + await session.refresh(merchant) + return merchant + + +@router.post( + "/{merchant_id}/logo/upload", + response_model=MerchantOut, + responses=NOT_FOUND, + summary="Logo hochladen", + description="Nimmt SVG, PNG oder JPEG bis 1 MB entgegen. Der Status wechselt auf " + "`manual`, die Markenfarbe wird aus der Datei ermittelt.", +) +async def upload_logo( + merchant_id: int, session: DbSession, file: UploadFile = File(description="Bilddatei.") +) -> Merchant: + merchant = await get_or_404(session, Merchant, merchant_id) + inhalt = await file.read() + await apply_upload(session, merchant, inhalt, file.content_type) + await session.commit() + await session.refresh(merchant) + return merchant + + +@router.post( + "/{merchant_id}/logo/resolve", + response_model=MerchantOut, + responses=NOT_FOUND, + summary="Logosuche erneut anstoßen", + description="Sucht und übernimmt den besten Treffer sofort. Mit `force=true` wird " + "auch eine manuelle Auswahl ersetzt.", +) +async def resolve_logo( + merchant_id: int, + session: DbSession, + force: bool = Query(default=False, description="Manuelle Auswahl überschreiben."), +) -> Merchant: + merchant = await get_or_404(session, Merchant, merchant_id) + await resolve_merchant_logo(session, merchant, force=force) + await session.commit() + await session.refresh(merchant) + return merchant + + +@router.delete( + "/{merchant_id}/logo", + response_model=MerchantOut, + responses=NOT_FOUND, + summary="Logo entfernen", + description="Löst die Zuordnung und setzt den Status zurück auf `pending`.", +) +async def remove_logo(merchant_id: int, session: DbSession) -> Merchant: + merchant = await get_or_404(session, Merchant, merchant_id) + merchant.logo_asset_id = None + merchant.logo_source = None + merchant.logo_status = LogoStatus.PENDING + merchant.brand_color = None + merchant.brand_color_dark = None + await session.commit() + await session.refresh(merchant) + return merchant diff --git a/backend/app/assets/simple_icons.json.gz b/backend/app/assets/simple_icons.json.gz new file mode 100644 index 0000000..021cb60 Binary files /dev/null and b/backend/app/assets/simple_icons.json.gz differ diff --git a/backend/app/core/config.py b/backend/app/core/config.py index e6f6d32..e77783c 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -62,6 +62,8 @@ class Settings(BaseSettings): logo_http_timeout_seconds: float = 5.0 logo_http_retries: int = 2 logo_max_upload_bytes: int = 1_048_576 + # Steuert, ob nach dem Anlegen einer Firma automatisch im Hintergrund gesucht wird. + logo_auto_resolve: bool = True # --- Benachrichtigungen -------------------------------------------------- notifications_enabled: bool = True diff --git a/backend/app/schemas/merchant.py b/backend/app/schemas/merchant.py index 29219b8..6bd9d73 100644 --- a/backend/app/schemas/merchant.py +++ b/backend/app/schemas/merchant.py @@ -45,3 +45,30 @@ class MerchantOut(ApiModel): logo_source: LogoSource | None logo_status: LogoStatus created_at: datetime + + +class LogoCandidateOut(ApiModel): + """Ein zur Auswahl stehendes Logo. Die Datei liegt bereits im lokalen Cache.""" + + candidate_id: int = Field(description="ID für /api/logos/{id} und für die Auswahl.") + source: LogoSource + title: str + score: float = Field( + ge=0, le=1, description="Trefferwahrscheinlichkeit; der höchste Wert ist vorausgewählt." + ) + mime: str + width: int | None = None + height: int | None = None + brand_color: str | None = None + is_preselected: bool = Field(description="Genau ein Kandidat ist vorausgewählt.") + + +class LogoSearchOut(ApiModel): + """Ergebnis der Logosuche.""" + + merchant_id: int + candidates: list[LogoCandidateOut] + + +class LogoSelectRequest(InputModel): + candidate_id: int = Field(description="ID aus der Kandidatenliste.") diff --git a/backend/app/services/colors.py b/backend/app/services/colors.py new file mode 100644 index 0000000..227aecb --- /dev/null +++ b/backend/app/services/colors.py @@ -0,0 +1,267 @@ +"""Ermittlung der Markenfarbe aus einem Logo. + +Bei SVG wird die häufigste Nicht-Graustufe aus den `fill`-Attributen genommen, +bei Rastergrafiken entscheidet ein k-Means über die Pixel. Zusätzlich wird eine +aufgehellte Variante berechnet, die auf dem dunklen Hintergrund der Oberfläche +den WCAG-Kontrast von 4,5:1 erreicht. +""" + +import colorsys +import hashlib +import io +import logging +import re +from collections import Counter +from dataclasses import dataclass + +from PIL import Image, UnidentifiedImageError + +logger = logging.getLogger(__name__) + +# Hintergrund der dunklen Oberfläche – gegen ihn wird der Kontrast geprüft. +DARK_BACKGROUND = "#0f1115" +MIN_CONTRAST = 4.5 + +# Ab diesem Sättigungswert gilt eine Farbe nicht mehr als Graustufe. +MIN_SATURATION = 0.12 +# Cluster unterhalb dieses Anteils sind Ausreißer und werden verworfen. +MIN_CLUSTER_SHARE = 0.05 + +# Deterministische Ersatzfarben, wenn ein Logo rein grau ist. +FALLBACK_PALETTE = ( + "#ef4444", + "#f97316", + "#eab308", + "#22c55e", + "#14b8a6", + "#3b82f6", + "#6366f1", + "#a855f7", + "#ec4899", +) + +_FILL_PATTERN = re.compile(r'fill\s*[:=]\s*["\']?\s*(#[0-9a-fA-F]{3,8}|rgb\([^)]+\))', re.I) +_STOP_COLOR_PATTERN = re.compile(r'stop-color\s*[:=]\s*["\']?\s*(#[0-9a-fA-F]{3,8})', re.I) +_RGB_PATTERN = re.compile(r"rgb\(\s*(\d+)[,\s]+(\d+)[,\s]+(\d+)", re.I) + + +@dataclass(frozen=True, slots=True) +class BrandColors: + """Markenfarbe und die für dunkle Oberflächen aufgehellte Variante.""" + + color: str + color_dark: str + + +# --- Umrechnungen -------------------------------------------------------------- + + +def normalize_hex(value: str) -> str | None: + """Bringt eine Farbangabe auf `#rrggbb`. Ungültiges ergibt None.""" + text = value.strip() + if not text.startswith("#"): + text = f"#{text}" + digits = text[1:] + if len(digits) in (4, 8): # Alphakanal abschneiden + digits = digits[:3] if len(digits) == 4 else digits[:6] + if len(digits) == 3: + digits = "".join(char * 2 for char in digits) + if len(digits) != 6 or not all(char in "0123456789abcdefABCDEF" for char in digits): + return None + return f"#{digits.lower()}" + + +def hex_to_rgb(value: str) -> tuple[int, int, int]: + normalized = normalize_hex(value) or "#000000" + return tuple(int(normalized[index : index + 2], 16) for index in (1, 3, 5)) # type: ignore[return-value] + + +def rgb_to_hex(rgb: tuple[int, int, int]) -> str: + red, green, blue = (max(0, min(255, round(channel))) for channel in rgb) + return f"#{red:02x}{green:02x}{blue:02x}" + + +def relative_luminance(rgb: tuple[int, int, int]) -> float: + """Relative Leuchtdichte nach WCAG 2.1.""" + channels = [] + for value in rgb: + srgb = value / 255 + channels.append(srgb / 12.92 if srgb <= 0.04045 else ((srgb + 0.055) / 1.055) ** 2.4) + red, green, blue = channels + return 0.2126 * red + 0.7152 * green + 0.0722 * blue + + +def contrast_ratio(first: str, second: str) -> float: + """Kontrastverhältnis zweier Farben nach WCAG (1:1 bis 21:1).""" + light = relative_luminance(hex_to_rgb(first)) + dark = relative_luminance(hex_to_rgb(second)) + if light < dark: + light, dark = dark, light + return (light + 0.05) / (dark + 0.05) + + +def saturation_of(rgb: tuple[int, int, int]) -> float: + _, _, saturation = colorsys.rgb_to_hls(*(channel / 255 for channel in rgb)) + return saturation + + +def lightness_of(rgb: tuple[int, int, int]) -> float: + _, lightness, _ = colorsys.rgb_to_hls(*(channel / 255 for channel in rgb)) + return lightness + + +def is_grayscale(rgb: tuple[int, int, int]) -> bool: + return saturation_of(rgb) < MIN_SATURATION + + +def lighten_for_dark_background(color: str, background: str = DARK_BACKGROUND) -> str: + """Hellt eine Farbe im HSL-Raum auf, bis der Kontrast mindestens 4,5:1 beträgt.""" + normalized = normalize_hex(color) + if normalized is None: + return "#e2e8f0" + if contrast_ratio(normalized, background) >= MIN_CONTRAST: + return normalized + + hue, lightness, saturation = colorsys.rgb_to_hls( + *(channel / 255 for channel in hex_to_rgb(normalized)) + ) + step = 0.02 + while lightness < 0.98: + lightness = min(lightness + step, 0.98) + channels = colorsys.hls_to_rgb(hue, lightness, saturation) + candidate = rgb_to_hex(tuple(round(channel * 255) for channel in channels)) + if contrast_ratio(candidate, background) >= MIN_CONTRAST: + return candidate + # Selbst bei maximaler Helligkeit nicht erreichbar (sehr dunkler Farbton): neutral aufhellen. + return "#e2e8f0" + + +def deterministic_color(name: str) -> str: + """Feste Farbe aus dem Namens-Hash – gleicher Name ergibt immer dieselbe Farbe.""" + digest = hashlib.sha256(name.strip().lower().encode("utf-8")).digest() + return FALLBACK_PALETTE[digest[0] % len(FALLBACK_PALETTE)] + + +def brand_colors(color: str | None, *, fallback_name: str = "") -> BrandColors: + """Baut das Farbpaar; ohne brauchbare Farbe greift der Namens-Hash.""" + normalized = normalize_hex(color) if color else None + if normalized is None: + normalized = deterministic_color(fallback_name) + return BrandColors(color=normalized, color_dark=lighten_for_dark_background(normalized)) + + +# --- SVG ----------------------------------------------------------------------- + + +def color_from_svg(content: bytes) -> str | None: + """Häufigste Nicht-Graustufe aus `fill`-Attributen und Verlaufsstopps.""" + try: + text = content.decode("utf-8", errors="ignore") + except Exception: # pragma: no cover - decode mit errors="ignore" wirft nicht + return None + + counter: Counter[str] = Counter() + for match in _FILL_PATTERN.finditer(text): + value = match.group(1) + if value.lower().startswith("rgb("): + numbers = _RGB_PATTERN.match(value) + if numbers is None: + continue + value = rgb_to_hex(tuple(int(part) for part in numbers.groups())) # type: ignore[arg-type] + normalized = normalize_hex(value) + if normalized and not is_grayscale(hex_to_rgb(normalized)): + counter[normalized] += 1 + + for match in _STOP_COLOR_PATTERN.finditer(text): + normalized = normalize_hex(match.group(1)) + if normalized and not is_grayscale(hex_to_rgb(normalized)): + counter[normalized] += 1 + + if not counter: + return None + return counter.most_common(1)[0][0] + + +# --- Rastergrafiken ------------------------------------------------------------ + + +def _kmeans(pixels: list[tuple[int, int, int]], k: int = 4, iterations: int = 20): + """Schlanker k-Means über RGB-Tripel. Liefert (Zentrum, Anzahl) je Cluster.""" + if not pixels: + return [] + unique = list(dict.fromkeys(pixels)) + if len(unique) <= k: + counts = Counter(pixels) + return [(color, counts[color]) for color in unique] + + # Deterministische Startpunkte: gleichmäßig über die sortierten Farben verteilt. + ordered = sorted(unique, key=lambda rgb: (relative_luminance(rgb), rgb)) + centers = [ordered[round(index * (len(ordered) - 1) / (k - 1))] for index in range(k)] + + for _ in range(iterations): + buckets: list[list[tuple[int, int, int]]] = [[] for _ in centers] + for pixel in pixels: + best = min( + range(len(centers)), + key=lambda index: sum( + (pixel[channel] - centers[index][channel]) ** 2 for channel in range(3) + ), + ) + buckets[best].append(pixel) + + moved = False + for index, bucket in enumerate(buckets): + if not bucket: + continue + center = tuple(round(sum(p[c] for p in bucket) / len(bucket)) for c in range(3)) + if center != centers[index]: + centers[index] = center # type: ignore[call-overload] + moved = True + if not moved: + break + + result = [(centers[index], len(bucket)) for index, bucket in enumerate(buckets) if bucket] + return sorted(result, key=lambda item: item[1], reverse=True) + + +def color_from_raster(content: bytes) -> str | None: + """Dominante Farbe einer Rastergrafik über k-Means (k=4) auf 64×64 Pixeln.""" + try: + with Image.open(io.BytesIO(content)) as image: + image = image.convert("RGBA") + image.thumbnail((64, 64), Image.Resampling.LANCZOS) + pixels = [ + (red, green, blue) + for red, green, blue, alpha in image.getdata() + if alpha >= 128 # durchscheinende Ränder verfälschen die Farbe + ] + except (UnidentifiedImageError, OSError, ValueError): + logger.debug("Rastergrafik konnte nicht gelesen werden.", exc_info=True) + return None + + if not pixels: + return None + + clusters = _kmeans(pixels, k=4) + total = sum(count for _, count in clusters) + if not total: + return None + + # Nur Cluster mit nennenswertem Anteil, davon der sattteste; bei ähnlicher + # Sättigung gewinnt der hellere. + relevant = [ + (color, count) for color, count in clusters if count / total >= MIN_CLUSTER_SHARE + ] or clusters + farbig = [(color, count) for color, count in relevant if not is_grayscale(color)] + if not farbig: + return None + + best = max(farbig, key=lambda item: saturation_of(item[0]) * 2 + lightness_of(item[0])) + return rgb_to_hex(best[0]) + + +def extract_color(content: bytes, mime: str) -> str | None: + """Ermittelt die Markenfarbe passend zum Dateityp.""" + if mime == "image/svg+xml": + return color_from_svg(content) + return color_from_raster(content) diff --git a/backend/app/services/logos.py b/backend/app/services/logos.py new file mode 100644 index 0000000..56a1d72 --- /dev/null +++ b/backend/app/services/logos.py @@ -0,0 +1,782 @@ +"""Logo-Service: Provider-Kette, Cache und Markenfarbe. + +Die Kette arbeitet Provider der Reihe nach ab und sammelt Kandidaten ein. Sie +endet immer beim generierten Buchstaben-Avatar, damit jede Firma ein Bild hat. +Ausgeliefert werden Logos ausschließlich lokal über `/api/logos/{id}`; im +Seitenaufruf entsteht kein Zugriff nach außen. +""" + +import asyncio +import gzip +import hashlib +import io +import json +import logging +import re +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import Path +from typing import Protocol, runtime_checkable + +import httpx +from PIL import Image +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.clock import utcnow +from app.core.config import settings +from app.core.errors import ValidationError +from app.models import LogoAsset, Merchant +from app.models.enums import LogoSource, LogoStatus +from app.services.colors import brand_colors, deterministic_color, extract_color +from app.services.merchants import domain_from_name, normalize_name + +logger = logging.getLogger(__name__) + +ASSETS_DIR = Path(__file__).resolve().parent.parent / "assets" +SIMPLE_ICONS_FILE = ASSETS_DIR / "simple_icons.json.gz" + +SVG_MIME = "image/svg+xml" +ALLOWED_UPLOAD_MIMES = {SVG_MIME, "image/png", "image/jpeg"} +MAX_CANDIDATES = 5 +# Ab diesem Wert gilt ein Treffer als eindeutig; die automatische Suche bricht dann ab. +EXACT_MATCH_SCORE = 0.95 +# Auch heruntergeladene Bilder werden begrenzt – ein Logo ist nie groß. +MAX_DOWNLOAD_BYTES = 1_048_576 + +_DOMAIN_PATTERN = re.compile(r"^(?:https?://)?(?:www\.)?([a-z0-9.-]+\.[a-z]{2,})", re.I) + + +@dataclass(frozen=True, slots=True) +class LogoCandidate: + """Ein gefundenes Logo, noch nicht gespeichert.""" + + source: LogoSource + content: bytes + mime: str + title: str + score: float + """0..1 – je höher, desto sicherer der Treffer. Bestimmt die Vorauswahl.""" + brand_color: str | None = None + source_url: str | None = None + width: int | None = None + height: int | None = None + + @property + def sha256(self) -> str: + return hashlib.sha256(self.content).hexdigest() + + +@runtime_checkable +class LogoProvider(Protocol): + """Ein Anbieter von Logo-Kandidaten.""" + + source: LogoSource + + def is_available(self) -> bool: + """False, wenn etwa der API-Schlüssel fehlt – der Provider wird dann übersprungen.""" + ... + + async def search( + self, name: str, domain: str | None, client: httpx.AsyncClient | None + ) -> list[LogoCandidate]: ... + + +# --- Hilfsfunktionen ----------------------------------------------------------- + + +def extract_domain(value: str | None) -> str | None: + """Reduziert eine Eingabe wie `https://www.netflix.com/de/` auf `netflix.com`.""" + if not value: + return None + match = _DOMAIN_PATTERN.match(value.strip()) + return match.group(1).lower() if match else None + + +def detect_mime(content: bytes, declared: str | None = None) -> str | None: + """Bestimmt den Bildtyp allein aus dem Inhalt. + + Die vom Aufrufer gemeldete Kopfzeile wird bewusst ignoriert: sonst könnte eine + beliebige Datei als Bild hochgeladen und später als solches ausgeliefert werden. + Der Parameter bleibt für die Protokollierung erhalten. + """ + if content.startswith(b"\x89PNG\r\n\x1a\n"): + return "image/png" + if content.startswith(b"\xff\xd8\xff"): + return "image/jpeg" + # SVG ist Text und darf mit Kommentaren, BOM oder DOCTYPE beginnen. + kopf = content[:2048].lstrip().lstrip(b"\xef\xbb\xbf") + if kopf.startswith((b" tuple[int | None, int | None]: + """Bildmaße, soweit ermittelbar. SVG ist skalierbar und liefert nichts.""" + if mime == SVG_MIME: + return None, None + try: + with Image.open(io.BytesIO(content)) as image: + return image.width, image.height + except Exception: + return None, None + + +async def fetch(client: httpx.AsyncClient, url: str, **kwargs) -> httpx.Response | None: + """Holt eine URL mit Wiederholungen. Fehler werden protokolliert, nie geworfen.""" + for attempt in range(settings.logo_http_retries + 1): + try: + response = await client.get(url, **kwargs) + if response.status_code == 200: + return response + # Bei 4xx lohnt kein zweiter Versuch. + if 400 <= response.status_code < 500: + logger.debug("Logo-Provider antwortete %s für %s", response.status_code, url) + return None + logger.debug("Logo-Provider antwortete %s für %s", response.status_code, url) + except httpx.HTTPError as exc: + logger.debug("Logo-Abruf fehlgeschlagen (%s): %s", url, exc) + if attempt < settings.logo_http_retries: + await asyncio.sleep(0.2 * (attempt + 1)) + return None + + +def _candidate_from_response( + response: httpx.Response, source: LogoSource, title: str, score: float +) -> LogoCandidate | None: + """Baut aus einer Antwort einen Kandidaten, sofern der Inhalt ein Bild ist.""" + content = response.content + if not content or len(content) > MAX_DOWNLOAD_BYTES: + return None + mime = detect_mime(content, response.headers.get("content-type")) + if mime is None: + return None + width, height = image_size(content, mime) + return LogoCandidate( + source=source, + content=content, + mime=mime, + title=title, + score=score, + brand_color=extract_color(content, mime), + source_url=str(response.url), + width=width, + height=height, + ) + + +# --- Provider 1: simple-icons (lokal, offline) --------------------------------- + + +@dataclass(frozen=True, slots=True) +class SimpleIcon: + slug: str + title: str + hex_color: str + path: str + aliases: tuple[str, ...] = () + + def to_svg(self) -> bytes: + """Erzeugt ein eigenständiges SVG in der Markenfarbe.""" + return ( + '{_escape(self.title)}' + f'' + ).encode() + + +def _escape(text: str) -> str: + return text.replace("&", "&").replace("<", "<").replace(">", ">") + + +@dataclass +class SimpleIconIndex: + """Nachschlagewerk über Slug, Titel und Aliasnamen.""" + + version: str + icons: list[SimpleIcon] + by_key: dict[str, SimpleIcon] = field(default_factory=dict) + + def lookup(self, key: str) -> SimpleIcon | None: + return self.by_key.get(key) + + +@lru_cache(maxsize=1) +def simple_icon_index() -> SimpleIconIndex: + """Lädt den gevendorten Index einmalig in den Speicher.""" + if not SIMPLE_ICONS_FILE.exists(): + logger.warning( + "simple-icons-Index fehlt (%s). Anlegen mit: make vendor-icons", SIMPLE_ICONS_FILE + ) + return SimpleIconIndex(version="", icons=[]) + + payload = json.loads(gzip.decompress(SIMPLE_ICONS_FILE.read_bytes()).decode("utf-8")) + icons = [ + SimpleIcon( + slug=entry["s"], + title=entry["t"], + hex_color=entry["h"], + path=entry["p"], + aliases=tuple(entry.get("a", ())), + ) + for entry in payload["icons"] + ] + + index = SimpleIconIndex(version=payload.get("version", ""), icons=icons) + for icon in icons: + # Der erste Treffer gewinnt, damit Aliasnamen echte Titel nicht verdrängen. + for key in (normalize_name(icon.title), icon.slug): + index.by_key.setdefault(key, icon) + for icon in icons: + for alias in icon.aliases: + index.by_key.setdefault(normalize_name(alias), icon) + return index + + +class SimpleIconsProvider: + """Lokaler Markenkatalog. Braucht keinen Schlüssel und keinen Netzzugriff.""" + + source = LogoSource.SIMPLE_ICONS + + def is_available(self) -> bool: + return bool(simple_icon_index().icons) + + async def search( + self, name: str, domain: str | None, client: httpx.AsyncClient | None + ) -> list[LogoCandidate]: + index = simple_icon_index() + treffer: list[tuple[SimpleIcon, float]] = [] + + exakt = index.lookup(normalize_name(name)) + if exakt is not None: + treffer.append((exakt, 1.0)) + + if domain: + # "netflix.com" -> "netflix" + aus_domain = index.lookup(normalize_name(domain.rsplit(".", 1)[0])) + if aus_domain is not None and all(icon is not aus_domain for icon, _ in treffer): + treffer.append((aus_domain, 0.95)) + + if not treffer: + treffer.extend(self._fuzzy(index, name)) + + return [ + LogoCandidate( + source=self.source, + content=icon.to_svg(), + mime=SVG_MIME, + title=icon.title, + score=score, + brand_color=f"#{icon.hex_color.lower()}", + source_url=None, + ) + for icon, score in treffer[:MAX_CANDIDATES] + ] + + def _fuzzy(self, index: SimpleIconIndex, name: str) -> list[tuple[SimpleIcon, float]]: + """Teiltreffer, wenn der Name nicht exakt passt – etwa 'Netflix Standard'.""" + gesucht = normalize_name(name) + if len(gesucht) < 3: + return [] + + ergebnisse: list[tuple[SimpleIcon, float]] = [] + for key, icon in index.by_key.items(): + if len(key) < 3: + continue + if gesucht.startswith(key) or key.startswith(gesucht): + # Je größer die Überschneidung, desto sicherer der Treffer. + anteil = min(len(key), len(gesucht)) / max(len(key), len(gesucht)) + ergebnisse.append((icon, 0.5 + 0.35 * anteil)) + + ergebnisse.sort(key=lambda item: item[1], reverse=True) + gesehen: set[str] = set() + eindeutig = [] + for icon, score in ergebnisse: + if icon.slug in gesehen: + continue + gesehen.add(icon.slug) + eindeutig.append((icon, score)) + return eindeutig[:3] + + +# --- Provider 2: logo.dev ------------------------------------------------------ + + +class LogoDevProvider: + """https://img.logo.dev – Abruf über Domain oder Firmenname. + + Nur aktiv, wenn `LOGODEV_API_KEY` gesetzt ist. Der Schlüssel ist ein + veröffentlichbarer Token und wird als Query-Parameter übergeben. + """ + + source = LogoSource.LOGODEV + base_url = "https://img.logo.dev" + + def is_available(self) -> bool: + return bool(settings.logodev_api_key) + + async def search( + self, name: str, domain: str | None, client: httpx.AsyncClient | None + ) -> list[LogoCandidate]: + if client is None or not self.is_available(): + return [] + + params = { + "token": settings.logodev_api_key, + "size": "256", + "format": "png", + # Ohne Treffer liefert logo.dev sonst ein generiertes Monogramm – + # das erledigt der eigene Avatar besser. + "fallback": "404", + } + versuche: list[tuple[str, float]] = [] + if domain: + versuche.append((f"{self.base_url}/{domain}", 0.9)) + slug = normalize_name(name) + if slug: + versuche.append((f"{self.base_url}/name/{slug}", 0.7)) + + kandidaten = [] + for url, score in versuche: + response = await fetch(client, url, params=params) + if response is None: + continue + kandidat = _candidate_from_response(response, self.source, name, score) + if kandidat is not None: + kandidaten.append(kandidat) + return kandidaten + + +# --- Provider 3: Brandfetch ---------------------------------------------------- + + +class BrandfetchProvider: + """https://api.brandfetch.io/v2/brands/{domain} mit Bearer-Token. + + Nur aktiv, wenn `BRANDFETCH_API_KEY` gesetzt ist **und** eine Domain bekannt + ist – die Brand-API schlägt ausschließlich über Kennungen nach. + """ + + source = LogoSource.BRANDFETCH + base_url = "https://api.brandfetch.io/v2/brands" + + def is_available(self) -> bool: + return bool(settings.brandfetch_api_key) + + async def search( + self, name: str, domain: str | None, client: httpx.AsyncClient | None + ) -> list[LogoCandidate]: + if client is None or not self.is_available() or not domain: + return [] + + response = await fetch( + client, + f"{self.base_url}/{domain}", + headers={"Authorization": f"Bearer {settings.brandfetch_api_key}"}, + ) + if response is None: + return [] + + try: + payload = response.json() + except ValueError: + logger.debug("Brandfetch lieferte kein JSON für %s", domain) + return [] + + marke = payload.get("name") or name + farbe = self._primary_color(payload) + + kandidaten: list[LogoCandidate] = [] + for url, score in self._logo_urls(payload)[:2]: + bild = await fetch(client, url) + if bild is None: + continue + kandidat = _candidate_from_response(bild, self.source, marke, score) + if kandidat is None: + continue + # Die von Brandfetch gemeldete Markenfarbe ist verlässlicher als die Pixelanalyse. + kandidaten.append( + LogoCandidate(**{**kandidat.__dict__, "brand_color": farbe or kandidat.brand_color}) + if farbe + else kandidat + ) + return kandidaten + + def _logo_urls(self, payload: dict) -> list[tuple[str, float]]: + """Bevorzugt Icons vor Wortmarken und SVG vor Rastergrafiken.""" + gewichtung = {"icon": 0.9, "symbol": 0.85, "logo": 0.8, "other": 0.6} + formatrang = {"svg": 0.05, "png": 0.03, "webp": 0.01, "jpeg": 0.0} + + ergebnisse: list[tuple[str, float]] = [] + for eintrag in payload.get("logos") or []: + basis = gewichtung.get(eintrag.get("type", "other"), 0.6) + for datei in eintrag.get("formats") or []: + quelle = datei.get("src") + if not quelle: + continue + ergebnisse.append((quelle, basis + formatrang.get(datei.get("format", ""), 0.0))) + ergebnisse.sort(key=lambda item: item[1], reverse=True) + return ergebnisse + + def _primary_color(self, payload: dict) -> str | None: + farben = payload.get("colors") or [] + for eintrag in farben: + if eintrag.get("type") == "primary" and eintrag.get("hex"): + return eintrag["hex"] + return farben[0].get("hex") if farben else None + + +# --- Provider 4: Favicon ------------------------------------------------------- + + +class FaviconProvider: + """Letzter Netzversuch: das Favicon der Domain über den Google-Dienst.""" + + source = LogoSource.FAVICON + base_url = "https://www.google.com/s2/favicons" + + def is_available(self) -> bool: + return True + + async def search( + self, name: str, domain: str | None, client: httpx.AsyncClient | None + ) -> list[LogoCandidate]: + if client is None: + return [] + + ziel = domain or domain_from_name(name) + if not ziel: + return [] + + response = await fetch(client, self.base_url, params={"sz": "128", "domain": ziel}) + if response is None: + return [] + # Ein geratener Domainname ist deutlich unsicherer als ein hinterlegter. + score = 0.5 if domain else 0.3 + kandidat = _candidate_from_response(response, self.source, name, score) + if kandidat is None: + return [] + # Der Platzhalter des Dienstes ist eine graue Weltkugel – die wollen wir nicht. + if kandidat.brand_color is None and (kandidat.width or 0) <= 16: + return [] + return [kandidat] + + +# --- Provider 5: generierter Avatar -------------------------------------------- + + +def initials(name: str) -> str: + """Ein bis zwei Initialen aus dem Firmennamen. Sonderzeichen bleiben außen vor.""" + woerter = [ + gesaeubert + for wort in re.split(r"[\s\-_/]+", name.strip()) + if (gesaeubert := re.sub(r"[^0-9A-Za-zÄÖÜäöüß]", "", wort)) + ] + if not woerter: + return "?" + if len(woerter) == 1: + wort = woerter[0] + return (wort[:2] if len(wort) > 1 else wort).upper() + return (woerter[0][0] + woerter[1][0]).upper() + + +def generate_avatar(name: str) -> LogoCandidate: + """Buchstaben-Avatar mit fester Farbe aus dem Namens-Hash. Schlägt nie fehl.""" + farbe = deterministic_color(name) + zeichen = _escape(initials(name)) + schriftgroesse = 44 if len(zeichen) > 1 else 54 + + svg = ( + '' + f"{_escape(name)}" + f'' + f'{zeichen}' + "" + ).encode() + + return LogoCandidate( + source=LogoSource.GENERATED, + content=svg, + mime=SVG_MIME, + title=name, + score=0.1, + brand_color=farbe, + width=96, + height=96, + ) + + +class GeneratedAvatarProvider: + source = LogoSource.GENERATED + + def is_available(self) -> bool: + return True + + async def search( + self, name: str, domain: str | None, client: httpx.AsyncClient | None + ) -> list[LogoCandidate]: + return [generate_avatar(name)] + + +# --- Die Kette ----------------------------------------------------------------- + +PROVIDER_CHAIN: tuple[LogoProvider, ...] = ( + SimpleIconsProvider(), + LogoDevProvider(), + BrandfetchProvider(), + FaviconProvider(), + GeneratedAvatarProvider(), +) + + +def requires_network(provider: LogoProvider) -> bool: + """True für Provider, die einen HTTP-Client benötigen.""" + return provider.source not in (LogoSource.SIMPLE_ICONS, LogoSource.GENERATED) + + +async def search_candidates( + name: str, domain: str | None = None, *, stop_on_exact: bool = False +) -> list[LogoCandidate]: + """Arbeitet die Provider-Kette ab und liefert bis zu fünf Kandidaten. + + Nach sha256 dedupliziert und nach Trefferwahrscheinlichkeit sortiert. Der + generierte Avatar ist immer enthalten, sodass die Liste nie leer ist. + + Mit `stop_on_exact` bleiben die Netz-Provider außen vor, sobald ein + eindeutiger Offline-Treffer vorliegt. Die automatische Suche nutzt das: für + eine bekannte Marke muss weder ein fremder Dienst befragt noch die + Firmenliste nach außen gegeben werden. Der Auswahldialog ruft bewusst ohne + diese Abkürzung auf, damit dem Nutzer trotzdem Alternativen angeboten werden. + """ + ziel_domain = extract_domain(domain) + + client: httpx.AsyncClient | None = None + kandidaten: list[LogoCandidate] = [] + try: + for provider in PROVIDER_CHAIN: + if not provider.is_available(): + continue + + if requires_network(provider): + if stop_on_exact and any( + kandidat.score >= EXACT_MATCH_SCORE for kandidat in kandidaten + ): + continue + if client is None: + client = httpx.AsyncClient( + timeout=settings.logo_http_timeout_seconds, + follow_redirects=True, + headers={"User-Agent": f"moneyfy/{settings.app_version}"}, + ) + + try: + gefunden = await provider.search(name, ziel_domain, client) + except Exception: # kein Provider darf die Kette abbrechen + logger.exception("Logo-Provider %s ist fehlgeschlagen.", provider.source) + continue + kandidaten.extend(gefunden) + finally: + if client is not None: + await client.aclose() + + return _dedupe(kandidaten)[:MAX_CANDIDATES] + + +def _dedupe(kandidaten: list[LogoCandidate]) -> list[LogoCandidate]: + """Entfernt inhaltsgleiche Bilder und sortiert nach Trefferwahrscheinlichkeit.""" + beste: dict[str, LogoCandidate] = {} + for kandidat in kandidaten: + vorhanden = beste.get(kandidat.sha256) + if vorhanden is None or kandidat.score > vorhanden.score: + beste[kandidat.sha256] = kandidat + return sorted(beste.values(), key=lambda item: item.score, reverse=True) + + +# --- Speicherung --------------------------------------------------------------- + + +def storage_path(sha256: str, mime: str) -> Path: + endung = {"image/png": "png", "image/jpeg": "jpg", SVG_MIME: "svg"}.get(mime, "bin") + return Path(f"{sha256}.{endung}") + + +async def store_candidate(session: AsyncSession, kandidat: LogoCandidate) -> LogoAsset: + """Legt das Logo im Cache ab. Bereits bekannte Inhalte werden wiederverwendet.""" + sha256 = kandidat.sha256 + vorhanden = ( + await session.execute(select(LogoAsset).where(LogoAsset.sha256 == sha256)) + ).scalar_one_or_none() + + relativ = storage_path(sha256, kandidat.mime) + absolut = settings.logo_storage_dir / relativ + if not absolut.exists(): + absolut.parent.mkdir(parents=True, exist_ok=True) + absolut.write_bytes(kandidat.content) + + if vorhanden is not None: + return vorhanden + + asset = LogoAsset( + sha256=sha256, + mime=kandidat.mime, + width=kandidat.width, + height=kandidat.height, + file_path=str(relativ), + source_url=kandidat.source_url, + fetched_at=utcnow(), + ) + session.add(asset) + await session.flush() + return asset + + +def read_asset(asset: LogoAsset) -> bytes: + """Liest ein zwischengespeichertes Logo von der Platte.""" + pfad = settings.logo_storage_dir / asset.file_path + if not pfad.exists(): + raise FileNotFoundError(pfad) + return pfad.read_bytes() + + +# --- Zuordnung zur Firma ------------------------------------------------------- + + +def _apply(merchant: Merchant, kandidat: LogoCandidate, asset: LogoAsset) -> None: + """Setzt Logo, Quelle und Farbpaar an der Firma.""" + farben = brand_colors(kandidat.brand_color, fallback_name=merchant.name) + merchant.logo_asset_id = asset.id + merchant.logo_source = kandidat.source + merchant.brand_color = farben.color + merchant.brand_color_dark = farben.color_dark + + +async def resolve_merchant_logo( + session: AsyncSession, merchant: Merchant, *, force: bool = False +) -> LogoAsset | None: + """Sucht ein Logo und ordnet den besten Treffer zu. + + Eine manuelle Auswahl oder ein Upload wird nie überschrieben, außer `force` + ist gesetzt. Die Suche endet notfalls beim generierten Avatar und schlägt + daher nicht fehl. + """ + if merchant.logo_status is LogoStatus.MANUAL and not force: + logger.debug("Firma %s hat ein manuelles Logo – Suche übersprungen.", merchant.name) + return None + + kandidaten = await search_candidates(merchant.name, merchant.domain, stop_on_exact=True) + if not kandidaten: + merchant.logo_status = LogoStatus.FAILED + await session.flush() + return None + + bester = kandidaten[0] + asset = await store_candidate(session, bester) + _apply(merchant, bester, asset) + merchant.logo_status = LogoStatus.RESOLVED + await session.flush() + return asset + + +async def resolve_merchant_logo_task(merchant_id: int) -> None: + """Hintergrundaufgabe mit eigener Session – wird nach der Antwort ausgeführt.""" + from app.db.session import SessionLocal + + try: + async with SessionLocal() as session: + merchant = await session.get(Merchant, merchant_id) + if merchant is None: + return + await resolve_merchant_logo(session, merchant) + await session.commit() + except Exception: + logger.exception("Logosuche für Firma %s fehlgeschlagen.", merchant_id) + + +async def apply_upload( + session: AsyncSession, merchant: Merchant, content: bytes, declared_mime: str | None +) -> LogoAsset: + """Übernimmt ein hochgeladenes Logo und sperrt es gegen die automatische Suche.""" + if len(content) > settings.logo_max_upload_bytes: + grenze = settings.logo_max_upload_bytes // 1024 + raise ValidationError( + f"Die Datei ist zu groß. Erlaubt sind höchstens {grenze} KB.", + code="logo_too_large", + ) + + mime = detect_mime(content, declared_mime) + if mime not in ALLOWED_UPLOAD_MIMES: + raise ValidationError( + "Nur SVG, PNG und JPEG werden unterstützt.", code="unsupported_media_type" + ) + + width, height = image_size(content, mime) + kandidat = LogoCandidate( + source=LogoSource.UPLOAD, + content=content, + mime=mime, + title=merchant.name, + score=1.0, + brand_color=extract_color(content, mime), + width=width, + height=height, + ) + + asset = await store_candidate(session, kandidat) + _apply(merchant, kandidat, asset) + merchant.logo_status = LogoStatus.MANUAL + await session.flush() + return asset + + +async def select_candidate(session: AsyncSession, merchant: Merchant, asset_id: int) -> LogoAsset: + """Übernimmt einen zuvor gefundenen Kandidaten als Logo der Firma.""" + asset = await session.get(LogoAsset, asset_id) + if asset is None: + raise ValidationError( + f"Das Logo mit der ID {asset_id} ist nicht mehr verfügbar. Bitte erneut suchen.", + code="candidate_not_found", + ) + + try: + content = read_asset(asset) + except FileNotFoundError as exc: + raise ValidationError( + "Die Logodatei fehlt im Cache. Bitte erneut suchen.", code="candidate_not_found" + ) from exc + + farbe = extract_color(content, asset.mime) + farben = brand_colors(farbe, fallback_name=merchant.name) + merchant.logo_asset_id = asset.id + merchant.brand_color = farben.color + merchant.brand_color_dark = farben.color_dark + # Eine bewusste Auswahl ist manuell und wird von der Automatik nicht angetastet. + merchant.logo_status = LogoStatus.MANUAL + await session.flush() + return asset + + +def cache_headers(asset: LogoAsset) -> dict[str, str]: + """Der Dateiname ist der Inhaltshash – der Inhalt kann sich nie ändern.""" + return { + "Cache-Control": "public, max-age=31536000, immutable", + "ETag": f'"{asset.sha256}"', + } + + +__all__ = [ + "LogoCandidate", + "apply_upload", + "cache_headers", + "generate_avatar", + "read_asset", + "resolve_merchant_logo", + "resolve_merchant_logo_task", + "search_candidates", + "select_candidate", + "simple_icon_index", + "store_candidate", +] diff --git a/backend/scripts/vendor_simple_icons.py b/backend/scripts/vendor_simple_icons.py new file mode 100644 index 0000000..438071a --- /dev/null +++ b/backend/scripts/vendor_simple_icons.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Erzeugt den lokalen simple-icons-Index aus dem npm-Paket. + +Das npm-Paket enthält rund 3.500 einzelne SVG-Dateien (etwa 15 MB). Für den +Betrieb genügen Slug, Titel, Markenfarbe, Pfaddaten und Aliasnamen – daraus wird +eine einzelne gzip-komprimierte JSON-Datei von etwa 2 MB. + +Aufruf: python scripts/vendor_simple_icons.py [--version 16] [--keep-tmp] +""" + +import argparse +import gzip +import json +import re +import shutil +import subprocess +import sys +import tarfile +import tempfile +from pathlib import Path + +TARGET = Path(__file__).resolve().parent.parent / "app" / "assets" / "simple_icons.json.gz" +PATH_PATTERN = re.compile(r' Path: + """Lädt das npm-Paket herunter und entpackt es.""" + result = subprocess.run( + ["npm", "pack", f"simple-icons@{version}"], + cwd=workdir, + capture_output=True, + text=True, + check=True, + ) + archive = workdir / result.stdout.strip().splitlines()[-1] + with tarfile.open(archive) as tar: + tar.extractall(workdir, filter="data") + return workdir / "package" + + +def collect_aliases(entry: dict) -> list[str]: + """Sammelt alternative Schreibweisen aus den Metadaten.""" + names: list[str] = [] + aliases = entry.get("aliases") or {} + for key in ("aka", "alt", "old", "dup"): + value = aliases.get(key) + if isinstance(value, list): + names.extend(item if isinstance(item, str) else item.get("title", "") for item in value) + elif isinstance(value, dict): + names.extend(str(item) for item in value.values()) + return [name for name in names if name] + + +def build(package: Path) -> list[dict]: + """Baut den kompakten Index aus Metadaten und SVG-Pfaden.""" + metadata = json.loads((package / "data" / "simple-icons.json").read_text(encoding="utf-8")) + icons: list[dict] = [] + + for entry in metadata: + svg_file = package / "icons" / f"{entry['slug']}.svg" + if not svg_file.exists(): + print(f" übersprungen (keine SVG-Datei): {entry['slug']}", file=sys.stderr) + continue + match = PATH_PATTERN.search(svg_file.read_text(encoding="utf-8")) + if not match: + print(f" übersprungen (kein Pfad): {entry['slug']}", file=sys.stderr) + continue + + icon = { + "s": entry["slug"], + "t": entry["title"], + "h": entry["hex"], + "p": match.group(1), + } + aliases = collect_aliases(entry) + if aliases: + icon["a"] = aliases + icons.append(icon) + + return icons + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--version", default="16", help="npm-Version von simple-icons.") + parser.add_argument("--keep-tmp", action="store_true", help="Arbeitsverzeichnis behalten.") + args = parser.parse_args() + + workdir = Path(tempfile.mkdtemp(prefix="simple-icons-")) + try: + print(f"Lade simple-icons@{args.version} …") + package = download(args.version, workdir) + version = json.loads((package / "package.json").read_text())["version"] + + icons = build(package) + payload = {"version": version, "icons": icons} + raw = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + TARGET.parent.mkdir(parents=True, exist_ok=True) + TARGET.write_bytes(gzip.compress(raw, 9)) + + print( + f"{len(icons)} Icons aus simple-icons {version} geschrieben nach " + f"{TARGET.relative_to(TARGET.parent.parent.parent)} " + f"({TARGET.stat().st_size / 1_000_000:.2f} MB)." + ) + return 0 + finally: + if not args.keep_tmp: + shutil.rmtree(workdir, ignore_errors=True) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 0084807..f21c073 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,6 +1,7 @@ """Gemeinsame Test-Fixtures. Nutzt eine separate Testdatenbank.""" import os +import tempfile # Muss vor dem ersten Import der Anwendung gesetzt sein, da die Settings gecacht werden. os.environ.setdefault( @@ -12,6 +13,9 @@ os.environ.setdefault( ) os.environ.setdefault("SECRET_KEY", "test-secret-key-mindestens-32-zeichen-lang") os.environ.setdefault("ENVIRONMENT", "test") +# Die Hintergrundsuche liefe außerhalb der Testtransaktion – sie wird gezielt aufgerufen. +os.environ.setdefault("LOGO_AUTO_RESOLVE", "false") +os.environ.setdefault("LOGO_STORAGE_DIR", tempfile.mkdtemp(prefix="moneyfy-logos-")) from collections.abc import AsyncGenerator from datetime import date diff --git a/backend/tests/test_colors.py b/backend/tests/test_colors.py new file mode 100644 index 0000000..9656b28 --- /dev/null +++ b/backend/tests/test_colors.py @@ -0,0 +1,158 @@ +"""Tests der Markenfarben-Ermittlung.""" + +import io + +import pytest +from PIL import Image + +from app.services.colors import ( + DARK_BACKGROUND, + MIN_CONTRAST, + brand_colors, + color_from_raster, + color_from_svg, + contrast_ratio, + deterministic_color, + hex_to_rgb, + is_grayscale, + lighten_for_dark_background, + normalize_hex, +) + + +def png(pixels: list[tuple[int, int, int, int]], size: tuple[int, int] = (8, 8)) -> bytes: + """Baut ein kleines PNG aus einer Pixelliste.""" + image = Image.new("RGBA", size) + image.putdata([pixels[index % len(pixels)] for index in range(size[0] * size[1])]) + buffer = io.BytesIO() + image.save(buffer, format="PNG") + return buffer.getvalue() + + +# --- Umrechnungen -------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("eingabe", "erwartet"), + [ + ("#E50914", "#e50914"), + ("E50914", "#e50914"), + ("#f00", "#ff0000"), + ("#E50914FF", "#e50914"), + ("kein-hex", None), + ("#12345", None), + ], +) +def test_hexwerte_werden_normalisiert(eingabe: str, erwartet: str | None) -> None: + assert normalize_hex(eingabe) == erwartet + + +def test_kontrastverhaeltnis_nach_wcag() -> None: + # Schwarz auf Weiß ist das Maximum von 21:1. + assert round(contrast_ratio("#000000", "#ffffff"), 1) == 21.0 + assert round(contrast_ratio("#ffffff", "#ffffff"), 1) == 1.0 + + +def test_graustufen_werden_erkannt() -> None: + assert is_grayscale(hex_to_rgb("#808080")) is True + assert is_grayscale(hex_to_rgb("#ffffff")) is True + assert is_grayscale(hex_to_rgb("#e50914")) is False + + +# --- Aufhellung ---------------------------------------------------------------- + + +@pytest.mark.parametrize( + "farbe", + ["#e50914", "#1d4ed8", "#166534", "#7c2d12", "#0058a3", "#e20074", "#312e81"], +) +def test_aufgehellte_farbe_erreicht_den_mindestkontrast(farbe: str) -> None: + aufgehellt = lighten_for_dark_background(farbe) + + assert contrast_ratio(aufgehellt, DARK_BACKGROUND) >= MIN_CONTRAST + + +def test_bereits_helle_farbe_bleibt_unveraendert() -> None: + # Ein helles Grün erfüllt den Kontrast bereits. + assert lighten_for_dark_background("#1ed760") == "#1ed760" + + +def test_farbpaar_faellt_auf_den_namenshash_zurueck() -> None: + ohne = brand_colors(None, fallback_name="Knuspertronic") + nochmal = brand_colors(None, fallback_name="Knuspertronic") + + assert ohne.color == nochmal.color # deterministisch + assert ohne.color == deterministic_color("Knuspertronic") + assert contrast_ratio(ohne.color_dark, DARK_BACKGROUND) >= MIN_CONTRAST + + +def test_namensfarbe_ignoriert_gross_schreibung_und_leerzeichen() -> None: + assert deterministic_color(" Netflix ") == deterministic_color("netflix") + + +# --- SVG ----------------------------------------------------------------------- + + +def test_svg_haeufigste_nicht_graustufe_gewinnt() -> None: + svg = ( + b'' + b'' + ) + + assert color_from_svg(svg) == "#e50914" + + +def test_svg_versteht_style_und_rgb_schreibweise() -> None: + svg = b'' + + assert color_from_svg(svg) == "#1db954" + + +def test_svg_beruecksichtigt_verlaufsstopps() -> None: + svg = b'' + + assert color_from_svg(svg) == "#833ab4" + + +def test_svg_ohne_farbe_liefert_nichts() -> None: + assert color_from_svg(b'') is None + assert color_from_svg(b'') is None + + +# --- Rastergrafiken ------------------------------------------------------------ + + +def test_raster_findet_die_dominante_farbe() -> None: + # Überwiegend Rot mit weißem Rand. + bild = png([(229, 9, 20, 255)] * 12 + [(255, 255, 255, 255)] * 4) + + ergebnis = color_from_raster(bild) + + assert ergebnis is not None + rot, gruen, blau = hex_to_rgb(ergebnis) + assert rot > 200 and gruen < 60 and blau < 60 + + +def test_raster_ignoriert_durchsichtige_pixel() -> None: + """Ein grüner Bildinhalt auf vollständig durchsichtigem Blau bleibt grün.""" + bild = png([(30, 215, 96, 255)] * 8 + [(0, 0, 255, 0)] * 8) + + ergebnis = color_from_raster(bild) + + assert ergebnis is not None + _, gruen, blau = hex_to_rgb(ergebnis) + assert gruen > 150 and blau < 150 + + +def test_reines_graustufenbild_liefert_keine_markenfarbe() -> None: + bild = png([(20, 20, 20, 255)] * 8 + [(240, 240, 240, 255)] * 8) + + assert color_from_raster(bild) is None + + +def test_vollstaendig_durchsichtiges_bild_liefert_nichts() -> None: + assert color_from_raster(png([(255, 0, 0, 0)])) is None + + +def test_kaputte_datei_wirft_nicht() -> None: + assert color_from_raster(b"kein bild") is None diff --git a/backend/tests/test_logos.py b/backend/tests/test_logos.py new file mode 100644 index 0000000..dff2fb0 --- /dev/null +++ b/backend/tests/test_logos.py @@ -0,0 +1,424 @@ +"""Tests der Provider-Kette, des Caches und der Logo-Endpunkte.""" + +import io +from pathlib import Path + +import httpx +import pytest +from httpx import AsyncClient +from PIL import Image +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import settings +from app.models import LogoAsset, Merchant +from app.models.enums import LogoSource, LogoStatus +from app.services.logos import ( + detect_mime, + extract_domain, + generate_avatar, + initials, + resolve_merchant_logo, + search_candidates, + simple_icon_index, + store_candidate, +) + + +@pytest.fixture +def kein_netz(monkeypatch: pytest.MonkeyPatch) -> None: + """Jeder ausgehende Request lässt den Test scheitern.""" + + class Verboten(httpx.AsyncClient): + def __init__(self, *args, **kwargs): + raise AssertionError("Es wurde ein ausgehender Request versucht.") + + monkeypatch.setattr(httpx, "AsyncClient", Verboten) + + +@pytest.fixture +def netz_ohne_treffer(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Ersetzt ausgehende Requests durch 404 und protokolliert die URLs.""" + aufgerufen: list[str] = [] + + class Leer: + def __init__(self, *args, **kwargs): + pass + + async def get(self, url, **kwargs): + aufgerufen.append(str(url)) + return httpx.Response(404, request=httpx.Request("GET", url)) + + async def aclose(self) -> None: + return None + + monkeypatch.setattr(httpx, "AsyncClient", Leer) + return aufgerufen + + +def png_bytes(farbe: tuple[int, int, int] = (229, 9, 20), size: int = 32) -> bytes: + image = Image.new("RGBA", (size, size), (*farbe, 255)) + buffer = io.BytesIO() + image.save(buffer, format="PNG") + return buffer.getvalue() + + +async def firma(session: AsyncSession, name: str, domain: str | None = None) -> Merchant: + from app.services.merchants import normalize_name + + merchant = Merchant(name=name, normalized_name=normalize_name(name), domain=domain) + session.add(merchant) + await session.flush() + return merchant + + +# --- simple-icons -------------------------------------------------------------- + + +def test_index_ist_vollstaendig_gevendort() -> None: + index = simple_icon_index() + + assert index.version, "Der Index muss eine Version melden." + assert len(index.icons) > 3000 + assert index.lookup("netflix") is not None + assert index.lookup("netflix").hex_color.lower() == "e50914" + + +async def test_netflix_ohne_schluessel_und_ohne_netz(kein_netz: None) -> None: + """Akzeptanzkriterium: bekannte Marke ohne API-Keys und ohne Netzzugriff.""" + kandidaten = await search_candidates("Netflix", stop_on_exact=True) + + bester = kandidaten[0] + assert bester.source is LogoSource.SIMPLE_ICONS + assert bester.mime == "image/svg+xml" + assert bester.brand_color == "#e50914" + assert bester.score == 1.0 + assert b" None: + ueber_domain = await search_candidates("Irgendwas", "spotify.com", stop_on_exact=True) + assert ueber_domain[0].title == "Spotify" + + # Rechtsform im Namen darf den Treffer nicht verhindern. + mit_rechtsform = await search_candidates("Deutsche Telekom AG", stop_on_exact=True) + assert mit_rechtsform[0].title == "Deutsche Telekom" + + +async def test_teiltreffer_wird_niedriger_bewertet(netz_ohne_treffer: list[str]) -> None: + """'Netflix Standard' ist kein exakter Name, soll aber Netflix finden.""" + kandidaten = await search_candidates("Netflix Standard") + + treffer = [k for k in kandidaten if k.source is LogoSource.SIMPLE_ICONS] + assert treffer[0].title == "Netflix" + assert 0.5 <= treffer[0].score < 0.95 + + +# --- Generierter Avatar -------------------------------------------------------- + + +@pytest.mark.parametrize( + ("name", "erwartet"), + [ + ("Netflix", "NE"), + ("Deutsche Telekom", "DT"), + ("1&1 Telecom", "1T"), + ("X", "X"), + ("!!!", "?"), + ], +) +def test_initialen(name: str, erwartet: str) -> None: + assert initials(name) == erwartet + + +def test_avatar_ist_deterministisch() -> None: + erster = generate_avatar("Knuspertronic GmbH") + zweiter = generate_avatar("Knuspertronic GmbH") + + assert erster.sha256 == zweiter.sha256 + assert erster.source is LogoSource.GENERATED + assert b"KG" in erster.content + + +def test_avatar_maskiert_sonderzeichen() -> None: + avatar = generate_avatar("Meier & Söhne") + + assert b"&" in avatar.content + assert b"Meier & S\xc3\xb6hne" in avatar.content + + +async def test_fantasiefirma_endet_beim_avatar(netz_ohne_treffer: list[str]) -> None: + """Akzeptanzkriterium: unbekannte Firma endet fehlerfrei beim Avatar.""" + kandidaten = await search_candidates("Knuspertronic GmbH", stop_on_exact=True) + + assert len(kandidaten) == 1 + assert kandidaten[0].source is LogoSource.GENERATED + assert kandidaten[0].brand_color is not None + + +async def test_providerfehler_werden_nicht_durchgereicht( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Ein abstürzender Provider darf die Kette nicht beenden.""" + + class Kaputt: + def __init__(self, *args, **kwargs): + pass + + async def get(self, url, **kwargs): + raise httpx.ConnectError("Netz weg") + + async def aclose(self) -> None: + return None + + monkeypatch.setattr(httpx, "AsyncClient", Kaputt) + + kandidaten = await search_candidates("Voellig Unbekannt AG") + + assert kandidaten[-1].source is LogoSource.GENERATED + + +# --- Hilfsfunktionen ----------------------------------------------------------- + + +@pytest.mark.parametrize( + ("eingabe", "erwartet"), + [ + ("netflix.com", "netflix.com"), + ("https://www.netflix.com/de/", "netflix.com"), + ("http://sub.example.co.uk/pfad", "sub.example.co.uk"), + ("keine domain", None), + (None, None), + ], +) +def test_domain_wird_herausgeloest(eingabe: str | None, erwartet: str | None) -> None: + assert extract_domain(eingabe) == erwartet + + +def test_dateityp_wird_an_der_signatur_erkannt() -> None: + assert detect_mime(png_bytes()) == "image/png" + assert detect_mime(b'') == "image/svg+xml" + assert detect_mime(b"\xff\xd8\xff\xe0irgendwas") == "image/jpeg" + assert detect_mime(b'') == "image/svg+xml" + # Eine behauptete Kopfzeile darf keinen Bildtyp vortäuschen. + assert detect_mime(b"harmloser text", "image/png") is None + assert detect_mime(b"%PDF-1.7", "image/svg+xml") is None + + +# --- Cache --------------------------------------------------------------------- + + +async def test_gleicher_inhalt_wird_nur_einmal_gespeichert(session: AsyncSession) -> None: + kandidat = generate_avatar("Knuspertronic") + + erster = await store_candidate(session, kandidat) + zweiter = await store_candidate(session, kandidat) + + assert erster.id == zweiter.id + anzahl = (await session.execute(select(func.count()).select_from(LogoAsset))).scalar_one() + assert anzahl == 1 + assert (settings.logo_storage_dir / erster.file_path).exists() + assert Path(erster.file_path).stem == kandidat.sha256 + + +async def test_zuordnung_setzt_farbe_und_status(session: AsyncSession, kein_netz: None) -> None: + merchant = await firma(session, "Netflix") + + asset = await resolve_merchant_logo(session, merchant) + + assert asset is not None + assert merchant.logo_asset_id == asset.id + assert merchant.logo_source is LogoSource.SIMPLE_ICONS + assert merchant.logo_status is LogoStatus.RESOLVED + assert merchant.brand_color == "#e50914" + # Die aufgehellte Variante ist für den dunklen Hintergrund gedacht. + assert merchant.brand_color_dark != merchant.brand_color + + +async def test_zweite_zuordnung_erzeugt_kein_neues_asset( + session: AsyncSession, kein_netz: None +) -> None: + """Akzeptanzkriterium: der zweite Aufruf löst keinen ausgehenden Request aus.""" + merchant = await firma(session, "Netflix") + + erster = await resolve_merchant_logo(session, merchant) + zweiter = await resolve_merchant_logo(session, merchant) + + assert erster.id == zweiter.id + anzahl = (await session.execute(select(func.count()).select_from(LogoAsset))).scalar_one() + assert anzahl == 1 + + +async def test_manuelles_logo_wird_nicht_ueberschrieben( + session: AsyncSession, kein_netz: None +) -> None: + merchant = await firma(session, "Netflix") + merchant.logo_status = LogoStatus.MANUAL + merchant.logo_asset_id = None + + ergebnis = await resolve_merchant_logo(session, merchant) + + assert ergebnis is None + assert merchant.logo_asset_id is None + + # Mit force wird sie doch ersetzt. + erzwungen = await resolve_merchant_logo(session, merchant, force=True) + assert erzwungen is not None + + +# --- Endpunkte ----------------------------------------------------------------- + + +async def test_logo_wird_lokal_mit_cache_header_ausgeliefert( + auth_client: AsyncClient, session: AsyncSession, kein_netz: None +) -> None: + merchant = await firma(session, "Netflix") + asset = await resolve_merchant_logo(session, merchant) + + antwort = await auth_client.get(f"/api/logos/{asset.id}") + + assert antwort.status_code == 200 + assert antwort.headers["content-type"].startswith("image/svg+xml") + assert antwort.headers["cache-control"] == "public, max-age=31536000, immutable" + assert antwort.headers["etag"] == f'"{asset.sha256}"' + assert b" None: + antwort = await auth_client.get("/api/logos/999999") + + assert antwort.status_code == 404 + assert antwort.json()["code"] == "not_found" + + +async def test_suche_liefert_kandidaten_mit_vorauswahl( + auth_client: AsyncClient, netz_ohne_treffer: list[str] +) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json() + + ergebnis = (await auth_client.post(f"/api/merchants/{angelegt['id']}/logo/search")).json() + + kandidaten = ergebnis["candidates"] + assert 1 <= len(kandidaten) <= 5 + assert kandidaten[0]["source"] == "simple-icons" + assert kandidaten[0]["is_preselected"] is True + assert sum(1 for k in kandidaten if k["is_preselected"]) == 1 + assert kandidaten[-1]["source"] == "generated" + + # Jeder Kandidat liegt bereits im Cache und ist sofort abrufbar. + for kandidat in kandidaten: + bild = await auth_client.get(f"/api/logos/{kandidat['candidate_id']}") + assert bild.status_code == 200 + + # Die Auswahl der Firma wurde durch die Suche noch nicht verändert. + firma_danach = (await auth_client.get(f"/api/merchants/{angelegt['id']}")).json() + assert firma_danach["logo_asset_id"] is None + assert firma_danach["logo_status"] == "pending" + + +async def test_kandidat_auswaehlen_setzt_manuell( + auth_client: AsyncClient, netz_ohne_treffer: list[str] +) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json() + kandidaten = (await auth_client.post(f"/api/merchants/{angelegt['id']}/logo/search")).json()[ + "candidates" + ] + + gewaehlt = ( + await auth_client.post( + f"/api/merchants/{angelegt['id']}/logo/select", + json={"candidate_id": kandidaten[0]["candidate_id"]}, + ) + ).json() + + assert gewaehlt["logo_asset_id"] == kandidaten[0]["candidate_id"] + assert gewaehlt["logo_status"] == "manual" + assert gewaehlt["brand_color"] == "#e50914" + + +async def test_auswahl_eines_unbekannten_kandidaten(auth_client: AsyncClient) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json() + + antwort = await auth_client.post( + f"/api/merchants/{angelegt['id']}/logo/select", json={"candidate_id": 999999} + ) + + assert antwort.status_code == 422 + assert antwort.json()["code"] == "candidate_not_found" + + +async def test_suche_mit_eigener_domain( + auth_client: AsyncClient, netz_ohne_treffer: list[str] +) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Streamdienst"})).json() + + await auth_client.post( + f"/api/merchants/{angelegt['id']}/logo/search", params={"domain": "spotify.com"} + ) + + # Die Domain wird übernommen und führt zum Treffer über den Domainnamen. + danach = (await auth_client.get(f"/api/merchants/{angelegt['id']}")).json() + assert danach["domain"] == "spotify.com" + + +async def test_upload_setzt_manuell_und_farbe(auth_client: AsyncClient) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Hausverwaltung"})).json() + + antwort = await auth_client.post( + f"/api/merchants/{angelegt['id']}/logo/upload", + files={"file": ("logo.png", png_bytes((30, 215, 96)), "image/png")}, + ) + + assert antwort.status_code == 200 + body = antwort.json() + assert body["logo_status"] == "manual" + assert body["logo_source"] == "upload" + assert body["logo_asset_id"] is not None + assert body["brand_color"] is not None + + bild = await auth_client.get(f"/api/logos/{body['logo_asset_id']}") + assert bild.headers["content-type"].startswith("image/png") + + +async def test_upload_lehnt_fremde_dateitypen_ab(auth_client: AsyncClient) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Testfirma"})).json() + + antwort = await auth_client.post( + f"/api/merchants/{angelegt['id']}/logo/upload", + files={"file": ("schaedlich.pdf", b"%PDF-1.7 kein Bild", "application/pdf")}, + ) + + assert antwort.status_code == 422 + assert antwort.json()["code"] == "unsupported_media_type" + + +async def test_upload_lehnt_zu_grosse_dateien_ab(auth_client: AsyncClient) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Grossfirma"})).json() + zu_gross = b"" + b"x" * (settings.logo_max_upload_bytes + 1) + b"" + + antwort = await auth_client.post( + f"/api/merchants/{angelegt['id']}/logo/upload", + files={"file": ("gross.svg", zu_gross, "image/svg+xml")}, + ) + + assert antwort.status_code == 422 + assert antwort.json()["code"] == "logo_too_large" + + +async def test_logo_entfernen_setzt_zurueck( + auth_client: AsyncClient, netz_ohne_treffer: list[str] +) -> None: + angelegt = (await auth_client.post("/api/merchants", json={"name": "Netflix"})).json() + await auth_client.post(f"/api/merchants/{angelegt['id']}/logo/resolve") + + entfernt = (await auth_client.delete(f"/api/merchants/{angelegt['id']}/logo")).json() + + assert entfernt["logo_asset_id"] is None + assert entfernt["logo_status"] == "pending" + assert entfernt["brand_color"] is None