feat(logos): Provider-Kette, lokaler Cache und Markenfarben
- simple-icons als kompakter Index im Repository (3.459 Marken, 2 MB gzip),
erzeugt von scripts/vendor_simple_icons.py bzw. `make vendor-icons`
- logo.dev und Brandfetch als optionale Adapter, ohne Schlüssel übersprungen
- Favicon-Fallback und generierter Buchstaben-Avatar als Garantie
- Bei eindeutigem Offline-Treffer unterbleiben Anfragen nach außen komplett
- Cache im Dateisystem nach SHA-256, Auslieferung nur über /api/logos/{id}
mit immutable-Header und ETag
- Markenfarbe aus SVG-Fills bzw. per k-Means (k=4) über 64x64 Pixel, dazu eine
aufgehellte Variante mit mindestens 4,5:1 Kontrast auf dunklem Grund
- Kandidatensuche mit Vorauswahl, Auswahl, Upload und Zurücksetzen
- Bildtyp wird nur noch am Inhalt bestimmt, nicht an der gemeldeten Kopfzeile
- 60 neue Tests, insgesamt 208 grün
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
This commit is contained in:
moneyfy
+16
-1
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
```
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
Binary file not shown.
@@ -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
|
||||
|
||||
@@ -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.")
|
||||
|
||||
@@ -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)
|
||||
@@ -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"<svg", b"<?xml", b"<!--", b"<!DOCTYPE")) and b"<svg" in kopf:
|
||||
return SVG_MIME
|
||||
if declared:
|
||||
logger.debug("Unbekannter Bildinhalt trotz gemeldetem Typ %s.", declared)
|
||||
return None
|
||||
|
||||
|
||||
def image_size(content: bytes, mime: str) -> 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 (
|
||||
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" role="img" '
|
||||
f'fill="#{self.hex_color.lower()}"><title>{_escape(self.title)}</title>'
|
||||
f'<path d="{self.path}"/></svg>'
|
||||
).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 = (
|
||||
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 96 96" role="img">'
|
||||
f"<title>{_escape(name)}</title>"
|
||||
f'<rect width="96" height="96" rx="20" fill="{farbe}"/>'
|
||||
f'<text x="48" y="48" fill="#ffffff" font-size="{schriftgroesse}" '
|
||||
'font-family="system-ui,-apple-system,Segoe UI,Roboto,sans-serif" font-weight="600" '
|
||||
f'text-anchor="middle" dominant-baseline="central">{zeichen}</text>'
|
||||
"</svg>"
|
||||
).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",
|
||||
]
|
||||
@@ -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\s+d="([^"]+)"')
|
||||
|
||||
|
||||
def download(version: str, workdir: Path) -> 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())
|
||||
@@ -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
|
||||
|
||||
@@ -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'<svg><path fill="#ffffff" d="a"/><path fill="#e50914" d="b"/>'
|
||||
b'<path fill="#e50914" d="c"/><path fill="#111111" d="d"/></svg>'
|
||||
)
|
||||
|
||||
assert color_from_svg(svg) == "#e50914"
|
||||
|
||||
|
||||
def test_svg_versteht_style_und_rgb_schreibweise() -> None:
|
||||
svg = b'<svg><path style="fill:rgb(29, 185, 84)" d="a"/></svg>'
|
||||
|
||||
assert color_from_svg(svg) == "#1db954"
|
||||
|
||||
|
||||
def test_svg_beruecksichtigt_verlaufsstopps() -> None:
|
||||
svg = b'<svg><linearGradient><stop stop-color="#833ab4"/></linearGradient></svg>'
|
||||
|
||||
assert color_from_svg(svg) == "#833ab4"
|
||||
|
||||
|
||||
def test_svg_ohne_farbe_liefert_nichts() -> None:
|
||||
assert color_from_svg(b'<svg><path d="M0 0"/></svg>') is None
|
||||
assert color_from_svg(b'<svg><path fill="#333333" d="M0 0"/></svg>') 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
|
||||
@@ -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"<svg" in bester.content
|
||||
|
||||
|
||||
async def test_treffer_ueber_alias_und_domain(kein_netz: None) -> 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"<title>Meier & S\xc3\xb6hne</title>" 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'<svg xmlns="http://www.w3.org/2000/svg"></svg>') == "image/svg+xml"
|
||||
assert detect_mime(b"\xff\xd8\xff\xe0irgendwas") == "image/jpeg"
|
||||
assert detect_mime(b'<?xml version="1.0"?><svg viewBox="0 0 1 1"/>') == "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"<svg" in antwort.content
|
||||
|
||||
# Mit passendem ETag antwortet der Server ohne Nutzlast.
|
||||
erneut = await auth_client.get(
|
||||
f"/api/logos/{asset.id}", headers={"If-None-Match": antwort.headers["etag"]}
|
||||
)
|
||||
assert erneut.status_code == 304
|
||||
|
||||
|
||||
async def test_unbekanntes_logo_liefert_404(auth_client: AsyncClient) -> 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"<svg>" + b"x" * (settings.logo_max_upload_bytes + 1) + b"</svg>"
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user