moneyfyandClaude Opus 5
8edb69fe6a
- 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
783 lines
26 KiB
Python
783 lines
26 KiB
Python
"""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",
|
||
]
|