- Anmeldung über Argon2id und JWT in httpOnly-Cookies, Refresh mit echter Rotation über die neue Tabelle refresh_token - AuthProvider-Protokoll als Vorbereitung für OIDC, Administrator-Anlage beim Erststart mit erzwungenem Passwortwechsel - CRUD für Konten, Kategorien (zweistufiger Baum), Firmen, Recurrences, Preisversionen, Buchungen, Budgets, Vorlagen und Sparziele - Fälligkeiten mit Overlay-Logik: abrufen, bestätigen, auslassen, zurücksetzen - Kontosalden zum Stichtag, Monatsübersicht mit Plan-Ist-Vergleich - SECRET_KEY jetzt mindestens 32 Zeichen; Platzhalter in Produktion abgelehnt - 61 neue Integrationstests, insgesamt 148 grün Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
89 lines
2.4 KiB
Python
89 lines
2.4 KiB
Python
"""Firmenstammdaten: Normalisierung und Suche."""
|
||
|
||
import re
|
||
import unicodedata
|
||
|
||
from sqlalchemy import or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models import Merchant
|
||
|
||
# Umlaute und ß werden ausgeschrieben, damit "Müller" und "Mueller" zusammenfallen.
|
||
_TRANSLITERATIONS = {
|
||
"ä": "ae",
|
||
"ö": "oe",
|
||
"ü": "ue",
|
||
"ß": "ss",
|
||
"æ": "ae",
|
||
"ø": "oe",
|
||
"å": "aa",
|
||
}
|
||
|
||
# Rechtsformen und Zusätze, die für die Wiedererkennung einer Marke irrelevant sind.
|
||
_LEGAL_SUFFIXES = (
|
||
"gmbh co kg",
|
||
"gmbh",
|
||
"ag",
|
||
"kg",
|
||
"ohg",
|
||
"ug",
|
||
"se",
|
||
"ev",
|
||
"mbh",
|
||
"inc",
|
||
"llc",
|
||
"ltd",
|
||
"plc",
|
||
"sa",
|
||
"bv",
|
||
"nv",
|
||
)
|
||
|
||
|
||
def normalize_name(name: str) -> str:
|
||
"""Kleinschreibung ohne Sonderzeichen – Grundlage für Eindeutigkeit und Logosuche."""
|
||
text = name.strip().lower()
|
||
for source, target in _TRANSLITERATIONS.items():
|
||
text = text.replace(source, target)
|
||
# Verbleibende Akzente auflösen (é -> e).
|
||
text = unicodedata.normalize("NFKD", text)
|
||
text = "".join(char for char in text if not unicodedata.combining(char))
|
||
# Punkte fallen ersatzlos weg, damit "B.V." und "e.V." als ein Wort erkannt werden.
|
||
text = text.replace(".", "")
|
||
text = re.sub(r"[^a-z0-9]+", " ", text).strip()
|
||
|
||
for suffix in _LEGAL_SUFFIXES:
|
||
if text.endswith(f" {suffix}"):
|
||
text = text[: -len(suffix) - 1].strip()
|
||
break
|
||
|
||
return re.sub(r"\s+", "", text)
|
||
|
||
|
||
def domain_from_name(name: str) -> str | None:
|
||
"""Rät eine Domain aus dem Firmennamen – nur als letzter Anhaltspunkt gedacht."""
|
||
slug = normalize_name(name)
|
||
if not slug or len(slug) < 3:
|
||
return None
|
||
return f"{slug}.com"
|
||
|
||
|
||
async def find_by_normalized_name(session: AsyncSession, name: str) -> Merchant | None:
|
||
stmt = select(Merchant).where(Merchant.normalized_name == normalize_name(name))
|
||
return (await session.execute(stmt)).scalar_one_or_none()
|
||
|
||
|
||
def search_statement(query: str | None):
|
||
"""Abfrage über Name, normalisierten Namen und Domain."""
|
||
stmt = select(Merchant).order_by(Merchant.name)
|
||
if query:
|
||
pattern = f"%{query.strip().lower()}%"
|
||
stmt = stmt.where(
|
||
or_(
|
||
Merchant.normalized_name.like(f"%{normalize_name(query)}%"),
|
||
Merchant.name.ilike(pattern),
|
||
Merchant.domain.ilike(pattern),
|
||
)
|
||
)
|
||
return stmt
|